Делать асинхронные запросы синхронными

Question

У меня есть базовая асинхронная система запросов / ответов вне очереди, которую я хочу сделать синхронной. Запросы и ответы можно сопоставлять, помечая запросы уникальными идентификаторами, которые, в свою очередь, будут сопровождать соответствующие ответы.

Моя попытка сделать его синхронным использует два ConcurrentHashMap s: один, который отображается из идентификаторов в результаты, и один, который сопоставляется с теми же идентификаторами в CountDownLatch es. Код выполняется следующим образом при выполнении запроса:

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;
    }
}

И код для обработки входящего результата:

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();
}

Этот метод вызывается из потока, который считывает все входящие результаты из базовой системы (существует только один такой поток чтения).

Прежде всего, я не уверен в безопасности потоков, но для этого также нет необходимости использовать два HashMap (особенно потому, что идентификаторы никогда не используются повторно). Есть идеи по улучшению?

Answer 1

Новая попытка, вдохновленная этим ответом на похожий вопрос :

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();
    }
}

Теперь мне нужен только один HashMap из этих 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

Я думаю, что версия с CountDownLatch, вероятно, является лучшим решением.«Параллелизм Java на практике» (Брайан Гетц) на самом деле говорит об этом (я думаю, он называет это защелкой значения).По сути, это механизм одноразовой синхронизации для устанавливаемого значения.

Хотя такую ​​защелку значения можно реализовать с помощью механизма wait () - notifyAll (), использование CountDownLatch дает более простую реализацию.

Методы countDownLatch await () - countDown () обеспечивают правильное отношение до события.

Answer 3

Использование ConcurrentHashMap подразумевает, что вы должны доверять только методам, которые определены как атомарные в документации, например:

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

, поэтому, если вы планируете сохранить их, вам следует изменить свое использование, это по крайней мере гарантирует безопасность потоков ваших хеш-карт.

Помимо этого, просто создайте новыйExecutor для каждого запрошенного запроса, который возвращает сам результат, так что поток будет ждать его завершения, прежде чем продолжать свою работу: таким образом, вы получите все это синхронизированным образом.