java - Javaで非同期httpリクエストを待機中

Question

Jetty HTTP クライアントを使用して、非同期で約 50 の HTTP 呼び出しを行っています。コードは次のようになります。

List<Address> addresses = getAddresses();
final List<String> done = Collections.synchronizedList(new LinkedList<String>());
List<ContentExchange> requests;
for (Address address : addresses) {
    ContentExchange ce = new ContentExchange() {
        @Override
        protected void onResponseComplete() throws IOException {
            //handle response
            done.add("done");
        }
    }
    ce.setURL(createURL(address));
    requests.add(ce);
}
for (ContentExchange ce : requests) {
    httpClient.send(ce);
}

while (done.size() != addresses.size()) {
    Thread.yield();
}

System.out.println("All addresses processed");

アドレスに関するデータを返す残りのサービスを呼び出しています。私が期待しているのはこれです：

1. 50 回の非同期 (ノンブロッキング) http 呼び出しを行います。
2. スレッドは、50 個すべてが終了するまで待機します。

しかし、それは機能していません。while ループがなくても問題なく動作しますが、50 個すべてが完了するまで待つ必要があります。50個すべてが完了するまで待つ方法はありますか?

ExecutorService とマルチスレッド ソリューションについても知っていますが、ノンブロッキング IO を備えたシングル スレッド ソリューションが必要です。

Answer 1

を使用しjava.util.concurrent.CountDownLatchてこれを管理します。

Eclipse Jetty 8.1.10.v20130312 の Siege.javaテスト クラスの例:

final CountDownLatch latch = new CountDownLatch(concurrent);   

for (int i=0;i<concurrent;i++)
{
    ConcurrentExchange ex = new ConcurrentExchange(client,latch,uris,repeats);
    if (!ex.next()) // this executes the client.send()
    {
        latch.countDown(); // count down if client.send() was in error
    }
}

latch.await(); // wait for all ConcurrentExchange's to complete (or error out)

注: ConcurrentExchangeは、Siege.java 内のプライベート クラスです。

次に、HttpExchangeオブジェクトCountDownLatch.countDown()で、次のメソッドで呼び出しを使用します

• onConnectionFailed(Throwable x) -例
• onException(Throwable x) - 例
• onExpire() - 例
• onResponseComplete() - 例

すべての例で a を使用して、AtomicBoolean counted1 回だけカウントされるようにしていることに注意してください。

if (!counted.getAndSet(true)) // get the value, then set it to true
{
    // only get here if counted returned false. (and that will only happen once)
    latch.countDown(); // count down this exchange as being done.
}