使异步查询同步答案

【问题标题】：Make asynchronous queries synchronous使异步查询同步
【发布时间】：2010-07-31 21:03:49
【问题描述】：

我有一个基础的异步乱序查询/响应系统，我想使其同步。可以通过使用唯一 ID 标记查询来映射查询和响应，这些 ID 反过来将伴随相应的响应。

我尝试使其同步使用两个ConcurrentHashMaps：一个从 ID 映射到结果，另一个从相同的 ID 映射到 CountDownLatches。执行查询时代码如下：

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

从一个线程调用此方法，该线程从底层系统读取所有传入的结果（只有一个这样的读取器线程）。

首先，我不确定线程安全性，但似乎也没有必要为此使用两个HashMaps（特别是因为 ID 从未被重用）。有什么改进的想法吗？

【问题讨论】：

所以删除使用“latch”的代码，但它不应该在这个函数中定义。不过，我没有看到任何可以使这个异步的东西。
实际执行异步查询的是底层系统。我已将该方法重命名为“executeAsyncWithId”以澄清。
嘿，我建议你发布自己的答案作为答案，并接受！

标签： java multithreading concurrency

【解决方案1】：

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

现在我只需要这些ResultFutures 中的一个HashMap：

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

【讨论】：

这个答案应该被删除，并将你原来问题中的文字作为更新。
@James Black，提出并回答您自己的问题是完全可以接受的。 @hakos 至少应该接受他自己的答案:)
@Tim Bender - 好的，仔细观察后我发现这是一个很好的答案，我最初认为这只是对原始示例的修改，只使用一个 HashMap .

【解决方案2】：

ConcurrentHashMap 的使用意味着您应该只信任文档中定义为原子的方法，例如：

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

所以如果你打算保留它们，你应该改变你的用法，这至少可以保证你的 hashmap 的线程安全。

除此之外，只需为每个请求的查询创建一个新的Executor，它会返回结果本身，以便线程在继续其工作之前等待其完成：这样你就会以同步的方式拥有整个事情..

【讨论】：

【解决方案3】：

我认为带有 CountDownLatch 的版本可能是最好的解决方案。 “Java Concurrency in Practice”（Brian Goetz）实际上谈到了这一点（我认为他称之为值锁存器）。它本质上是一种对正在设置的值的一次性同步机制。

虽然可以使用 wait()-notifyAll() 机制实现这样的值锁存器，但使用 CountDownLatch 会产生更简单的实现。

CountDownLatch await()-countDown() 方法提供了正确的先发生关系。

【讨论】：