在风暴中使用 Apache 的 AsyncHttpClient答案

【问题标题】：using Apache's AsyncHttpClient in a storm bolt在风暴中使用 Apache 的 AsyncHttpClient
【发布时间】：2016-06-26 21:25:39
【问题描述】：

我有一个螺栓正在为每个元组进行 API 调用 (HTTP Get)。为了避免需要等待响应，我希望使用 apache HttpAsyncClient。

在bolt的prepare方法中实例化客户端后，execute方法从元组中构造URL并调用sendAsyncGetRequest(url)：

private void sendAsyncGetRequest(String url){

    httpclient.execute(new HttpGet(url), new FutureCallback<HttpResponse>() {

        @Override
        public void completed(HttpResponse response) {
            LOG.info("Response Code : " + response.getStatusLine());
            LOG.debug(response.toString());
        }

        @Override
        public void failed(Exception ex) {
            LOG.warn("Async http request failed!", ex);
        }

        @Override
        public void cancelled() {
            LOG.warn("Async http request canceled!");
        }
    });
}

拓扑已部署，但 Storm UI 显示错误：

java.lang.RuntimeException: java.lang.IllegalStateException: Request cannot be executed; I/O reactor status: STOPPED at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:12

【问题讨论】：

我觉得没有足够的数据来检查这个问题。 httpClient 是如何实例化的？你能发布完整的错误堆栈吗？

标签： java apache-storm asynchttpclient

【解决方案1】：

我让它正常工作。需要注意的关键事项是：

在螺栓类范围内声明客户端

public class MyRichBolt extends BaseRichBolt {
    private CloseableHttpAsyncClient httpclient;

在bolt的prepare方法中实例化和统计客户端

@Override
public final void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
    try {
        // start the http client
        httpclient = HttpAsyncClients.createDefault();
        httpclient.start();
        // other initialization code ... 
    } catch (Throwable exception) {
    // handle errors
    }
}

在螺栓的执行方法中调用

@Override
public final void execute(Tuple tuple) {
    // format the request url
    String url = ... 
    sendAsyncGetRequest(url);
}


private void sendAsyncGetRequest(String url){
    logger.debug("Async call to URL...");
    HttpGet request = new HttpGet(url);
    HttpAsyncRequestProducer producer = HttpAsyncMethods.create(request);
    AsyncCharConsumer<HttpResponse> consumer = new AsyncCharConsumer<HttpResponse>() {

        HttpResponse response;

        @Override
        protected void onResponseReceived(final HttpResponse response) {
        this.response = response;
        }

        @Override
        protected void onCharReceived(final CharBuffer buf, final IOControl ioctrl) throws IOException {
            // Do something useful
        }

        @Override
        protected void releaseResources() {
        }

        @Override
        protected HttpResponse buildResult(final HttpContext context) {
            return this.response;
        }
    };

    httpclient.execute(producer, consumer, new FutureCallback<HttpResponse>() {

        @Override
        public void completed(HttpResponse response) {
            // do something useful with the response
            logger.debug(response.toString());
        }

        @Override
        public void failed(Exception ex) {
            logger.warn("!!! Async http request failed!", ex);
        }

        @Override
        public void cancelled() {
            logger.warn("Async http request canceled!");
        }
    });
}

【讨论】：

解释根本原因会更有帮助。即，为什么您规定的解决方案有效？

【解决方案2】：

您是否在执行回调之前关闭了主流程中的客户端 (client.close();)？

错误是说IO路径已经关闭。一般来说，异步客户端的实例应该重新用于重复请求，并且只有在发出“所有”请求时才销毁，例如在应用程序关闭时。

【讨论】：