如何在 Apache 异步客户端中配置允许的待处理请求数答案

【问题标题】：How to configure the number of allowed pending requests in Apache async client如何在 Apache 异步客户端中配置允许的待处理请求数
【发布时间】：2015-05-07 13:02:07
【问题描述】：

我正在开发一个 HTTP 客户端应用程序，使用 Apache httpasyncclient 版本 4.0.2。

我想配置待处理请求的最大数量。最初我假设这个数字与最大连接数相同。我通过以下方式将其设置为 20：

    final CloseableHttpAsyncClient httpclient;

在构造函数中：

    final NHttpConnectionFactory<ManagedNHttpClientConnection> connFactory = new ManagedNHttpClientConnectionFactory(new DefaultHttpRequestWriterFactory(), new DefaultHttpResponseParserFactory(), HeapByteBufferAllocator.INSTANCE);
    final IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
            .setIoThreadCount(4)
            .setConnectTimeout(30000)
            .setSoTimeout(30000)
            .build();
    final PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(ioReactorConfig), connFactory);
    final int maxConns = 20;
    connManager.setDefaultMaxPerRoute(maxConns);
    connManager.setMaxTotal(maxConns);
    httpclient = HttpAsyncClientBuilder.create().setConnectionManager(connManager).build();
    httpclient.start();

及以后：

    final BasicAsyncRequestProducer requestProducer = new BasicAsyncRequestProducer(URIUtils.extractHost(URI.create(serverAddress)), request) {
        @Override
        public void requestCompleted(HttpContext context) {
            pendings.add(callback);
            logMessage(Direction.REQUEST, req);
            handler.onContentWriteCompleted();
        }
    };
    httpclient.execute(requestProducer, HttpAsyncMethods.createConsumer(), new HttpClientContext(), callback);

回调是我处理响应的地方。

作为概念证明，它失败了。确实 get do 4 个线程在运行，但是当我尝试同时发送 20 条消息时，只有 8 条立即发送，其余的必须等待服务器响应。

Apache 调试消息表明确实已创建 20 个连接。看来还得做更多的配置。

【问题讨论】：

标签： java apache httpclient apache-httpclient-4.x

【解决方案1】：

Apache HttpAsyncClient 维护一个无限制的请求执行队列，并且不会尝试限制挂起请求的数量。各种应用程序可能希望或可能不希望限制请求率，并且没有简单的方法可以满足所有要求。

然而，使用一个简单的信号量可以很容易地限制并发请求的数量。

final Semaphore semaphore = new Semaphore(maxConcurrencyLevel);
for (int i = 0; i < n; i++) {

    semaphore.acquire();
    this.httpclient.execute(
            new BasicAsyncRequestProducer(target, request),
            new MyResponseConsumer(),
            new FutureCallback<HttpResponse>() {

                @Override
                public void completed(final HttpResponse result) {
                    semaphore.release();
                }

                @Override
                public void failed(final Exception ex) {
                    semaphore.release();
                }

                @Override
                public void cancelled() {
                    semaphore.release();
                }

            });
}

【讨论】：