Faire des requêtes asynchrones synchrones

Question

J'ai un système d'interrogation/réponse asynchrone sous-jacente asynchrone que je veux rendre synchrone. Les requêtes et les réponses peuvent être mappées en marquant les requêtes avec des identifiants uniques qui, à leur tour, accompagneront les réponses correspondantes.

Ma tentative pour la rendre synchrone utilise deux ConcurrentHashMap s: une qui mappe des ID aux résultats, et une qui mappe des mêmes ID à CountDownLatch es. Le code va comme ceci lors de l'exécution d'une requête:

public Result execute(Query query) throws InterruptedException { 
    int id = atomicInteger.incrementAndGet(); 
    CountDownLatch latch = new CountDownLatch(1); 
    latchMap.put(id, latch); 
    query.executeAsyncWithId(id); // Probably returns before result is ready 
    try { 
     latch.await(); // Blocks until result is ready 
     return resultMap.remove(id); 
    } catch (InterruptedException e) { 
     latchMap.remove(id); // We are not waiting anymore 
     throw e; 
    } 
} 

Et le code pour traiter un résultat entrant:

public void handleResult(Result result) { 
    int id = result.getId(); 
    CountDownLatch latch = latchMap.remove(id); 
    if (latch == null) { 
     return; // Nobody wants result 
    } 
    resultMap.put(id, result); 
    latch.countDown(); 
} 

Cette méthode est appelée à partir d'un fil qui lit tous les résultats entrants du système sous-jacent (il n'y a qu'un seul thread de lecteur de ce type). Tout d'abord, je ne suis pas sûr de la sécurité des threads, mais il semble inutile d'utiliser deux HashMap pour cela (en particulier parce que les ID ne sont jamais réutilisés). Des idées d'améliorations?

Answer 1

une nouvelle tentative inspirée par this answer to a similar question:

public class ResultFuture { 

    private volatile Result result = null; 
    private final CountDownLatch latch = new CountDownLatch(1); 

    public Result get() throws InterruptedException { 
     latch.await(); 
     return result; 
    } 

    public void set(Result result) { 
     this.result = result; 
     latch.countDown(); 
    } 
} 

maintenant je besoin que d'un HashMap de ces ResultFuture s:

public Result execute(Query query) throws InterruptedException { 
    int id = atomicInteger.incrementAndGet(); 
    ResultFuture resultFuture = new ResultFuture(); 
    resultFutureMap.put(id, resultFuture); 
    query.executeAsyncWithId(id); // Probably returns before result is ready 
    try { 
     return resultFuture.get(); // Blocks until result is ready 
    } finally { 
     resultFutureMap.remove(id); 
    } 
} 

public void handleResult(Result result) { 
    int id = result.getId(); 
    ResultFuture resultFuture = resultFutureMap.get(id); 
    if (resultFuture == null) { 
     return; // Nobody wants result 
    } 
    resultFuture.set(result); 
} 

Answer 2

Utilisation de ConcurrentHashMap implique que vous devez faire confiance à quelques méthodes qui sont définies comme atomique dans la documentation comme:

• putIfAbsent(K key, V val)
• replace(K key, V val)
• remove(K key, V val)

donc si vous prévoyez de garder Si vous devez changer votre utilisation, cela garantira au moins la sécurité des threads de vos hashmaps. En dehors de cela, il suffit de créer un nouveau Executor pour chaque requête demandée qui retourne le résultat lui-même afin que le thread attende son achèvement avant de poursuivre son travail: de cette façon, vous aurez le tout de manière synchronisée. .

Answer 3

Je pense que la version avec le CountDownLatch est probablement la meilleure solution. "Java Concurrency in Practice" (Brian Goetz) parle en fait de cela (je pense qu'il l'appelle le verrou de la valeur). Il s'agit essentiellement d'un mécanisme de synchronisation unique sur une valeur définie.

Bien qu'il soit possible d'implémenter un tel verrou de valeur avec le mécanisme wait() - notifyAll(), l'utilisation de CountDownLatch donne une implémentation plus simple.

Les méthodes countDownLatch await() countDown() fournissent une relation qui se produit avant l'événement.