异步httpClient(Async HttpClient)

2021-10-15 14:20:41

一、简介
二、mvn依赖
三、客户端
- 3.1 官网实例
- 3.2. 根据官方文档的介绍，简单封装了一个异步HttpClient工具类
- 3.3 基本原理
四、参考文档

一、简介

HttpClient提供了两种I/O模型：经典的java阻塞I/O模型和基于Java NIO的异步非阻塞事件驱动I/O模型。

Java中的阻塞I/O是一种高效、便捷的I/O模型，非常适合并发连接数量相对适中的高性能应用程序。只要并发连接的数量在1000个以下并且连接大多忙于传输数据，阻塞I/O模型就可以提供最佳的数据吞吐量性能。然而，对于连接大部分时间保持空闲的应用程序，上下文切换的开销可能会变得很大，这时非阻塞I/O模型可能会提供更好的替代方案。异步I/O模型可能更适合于比较看重资源高效利用、系统可伸缩性、以及可以同时支持更多HTTP连接的场景。

二、mvn依赖

httpclient在4.x之后开始提供基于nio的异步版本httpasyncclient,httpasyncclient借助了Java并发库和nio进行封装(虽说NIO是同步非阻塞IO,但是HttpAsyncClient提供了回调的机制,与netty类似,所以可以模拟类似于AIO的效果),其调用方式非常便捷.

<dependency>
  <groupId>org.apache.httpcomponents</groupId>
  <artifactId>httpasyncclient</artifactId>
  <version>4.1.4</version>
</dependency>

三、客户端

3.1 官网实例

CloseableHttpAsyncClient httpclient = HttpAsyncClients.createDefault();
try {
    // Start the client
    httpclient.start();

    // Execute request
    final HttpGet request1 = new HttpGet("http://www.apache.org/");
    Future<HttpResponse> future = httpclient.execute(request1, null);
    // and wait until a response is received
    HttpResponse response1 = future.get();
    System.out.println(request1.getRequestLine() + "->" + response1.getStatusLine());

    // One most likely would want to use a callback for operation result
    final CountDownLatch latch1 = new CountDownLatch(1);
    final HttpGet request2 = new HttpGet("http://www.apache.org/");
    httpclient.execute(request2, new FutureCallback<HttpResponse>() {

        public void completed(final HttpResponse response2) {
            latch1.countDown();
            System.out.println(request2.getRequestLine() + "->" + response2.getStatusLine());
        }

        public void failed(final Exception ex) {
            latch1.countDown();
            System.out.println(request2.getRequestLine() + "->" + ex);
        }

        public void cancelled() {
            latch1.countDown();
            System.out.println(request2.getRequestLine() + " cancelled");
        }

    });
    latch1.await();

    // In real world one most likely would also want to stream
    // request and response body content
    final CountDownLatch latch2 = new CountDownLatch(1);
    final HttpGet request3 = new HttpGet("http://www.apache.org/");
    HttpAsyncRequestProducer producer3 = HttpAsyncMethods.create(request3);
    AsyncCharConsumer<HttpResponse> consumer3 = 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(producer3, consumer3, new FutureCallback<HttpResponse>() {

        public void completed(final HttpResponse response3) {
            latch2.countDown();
            System.out.println(request3.getRequestLine() + "->" + response3.getStatusLine());
        }

        public void failed(final Exception ex) {
            latch2.countDown();
            System.out.println(request3.getRequestLine() + "->" + ex);
        }

        public void cancelled() {
            latch2.countDown();
            System.out.println(request3.getRequestLine() + " cancelled");
        }

    });
    latch2.await();

} finally {
    httpclient.close();
}

3.2. 根据官方文档的介绍，简单封装了一个异步HttpClient工具类

public class AsyncHttpClientUtil {
    private static final Logger logger = LoggerFactory.getLogger(AsyncHttpClientUtil.class);
    //从池中获取链接超时时间(ms)
    private static final int CONNECTION_REQUEST_TIMEOUT = 10000;
    //建立链接超时时间(ms)
    private static final int CONNECT_TIMEOUT = 10000;
    //读取超时时间(ms)
    private static final int SOCKET_TIMEOUT = 5000;
    //连接数
    private static final int MAX_TOTAL = 50;
    private static final int MAX_PER_ROUTE = 50;

    private static final CloseableHttpAsyncClient httpclient;
    private static PoolingNHttpClientConnectionManager poolManager;

    static {
        httpclient=init();
        httpclient.start();
    }

    private static CloseableHttpAsyncClient init() {
        CloseableHttpAsyncClient client=null;
        try {
            //配置io线程
            IOReactorConfig ioReactorConfig = IOReactorConfig.custom().
                    setIoThreadCount(Runtime.getRuntime().availableProcessors())
                    .setSoKeepAlive(true)
                    .build();
            //创建一个ioReactor
            ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
//            poolManager=new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor());
            poolManager=new PoolingNHttpClientConnectionManager(ioReactor);
            //设置连接池大小
            poolManager.setMaxTotal(MAX_TOTAL);
            poolManager.setDefaultMaxPerRoute(MAX_PER_ROUTE);
            // 配置请求的超时设置
            RequestConfig requestConfig = RequestConfig.custom()
                    .setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT)
                    .setConnectTimeout(CONNECT_TIMEOUT)
                    .setSocketTimeout(SOCKET_TIMEOUT)
                    .build();

            client= HttpAsyncClients.custom()
                    .setConnectionManager(poolManager)
                    .setDefaultRequestConfig(requestConfig)
                    .build();
            return client;
        } catch (IOReactorException e) {
            e.printStackTrace();
        }
        return client;
    }

    public static String get(String url, List<NameValuePair> ns) {
        HttpGet httpget;
        URIBuilder uri = new URIBuilder();
        try {
            if (ns!=null){
                uri.setPath(url);
                uri.addParameters(ns);
                httpget = new HttpGet(uri.build());
            }else{
                httpget = new HttpGet(url);
            }

            // One most likely would want to use a callback for operation result
            httpclient.execute(httpget, new FutureCallback<HttpResponse>() {

                public void completed(final HttpResponse response) {
                    System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
                    try {
                        System.out.println("当前请求状态："+poolManager.getTotalStats()+", response="+EntityUtils.toString(response.getEntity()));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                public void failed(final Exception ex) {
                    System.out.println(httpget.getRequestLine() + "->" + ex);
                }

                public void cancelled() {
                    System.out.println(httpget.getRequestLine() + " cancelled");
                }

            });

        }catch (Exception e){
            logger.error("[发送get请求失败]URL:{},异常:",uri.getUserInfo(), e);
        }
        return null;
    }

    public static void main(String[] args) {
        for (int i=0; i<10;i++){
            System.out.println("第" + i +"次：");
            get("http://httpbin.org/get",null);
        }
    }
}

3.3 基本原理

HTTPAysncClient 最后使用的是 InternalHttpAsyncClient，在 InternalHttpAsyncClient 中有个 ConnectionManager，这个就是我们管理连接的管理器。

而在 httpAsync 中只有一个实现那就是 PoolingNHttpClientConnectionManager。这个连接管理器中有两个我们比较关心的，一个是 Reactor，一个是 Cpool：

Reactor：所有的 Reactor 这里都是实现了 IOReactor 接口。在 PoolingNHttpClientConnectionManager 中会有拥有一个 Reactor，那就是 DefaultConnectingIOReactor，这个 DefaultConnectingIOReactor，负责处理 Acceptor。

在 DefaultConnectingIOReactor 有个 excutor 方法，生成 IOReactor 也就是我们图中的 BaseIOReactor，进行 IO 的操作。

CPool：在 PoolingNHttpClientConnectionManager 中有个 CPool，主要是负责控制我们连接，我们上面所说的 maxTotal 和 defaultMaxPerRoute，都是由其进行控制。

如果每个路由有满了，它会断开最老的一个链接;如果总共的 total 满了，它会放入 leased 队列，释放空间的时候就会将其重新连接。

关于Reactor可参考： https://www.cnblogs.com/doit8791/p/7461479.html

四、参考文档