Apache异步http工具HttpAsyncClient的使用

1. 起因

  Apache的HttpClient在4.X之后推出了异步http版本，项目突然要用到之下措手不及，并且犯了一些理解上的错误，这里记录一下。

2. 使用HttpAsyncClient的最简例子

首先在pom文件中导入（版本可能不一样，可以去http://mvnrepository.com/找）

<!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpcore-nio -->
<dependency>
  <groupId>org.apache.httpcomponents</groupId>
  <artifactId>httpcore-nio</artifactId>
  <version>4.4.10</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.httpcomponents/httpasyncclient -->
<dependency>
  <groupId>org.apache.httpcomponents</groupId>
  <artifactId>httpasyncclient</artifactId>
  <version>4.1.4</version>
</dependency>

最简单的一个使用：

public class TestAsyncClient {
    public static void main(String[] args) {
        CloseableHttpAsyncClient httpAsyncClient = HttpAsyncClients.createDefault();
        final HttpGet request = new HttpGet("http://www.baidu.com");
        httpAsyncClient.start();
        CountDownLatch latch = new CountDownLatch(1);
        httpAsyncClient.execute(request, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(final HttpResponse response2) {
                latch.countDown();
                System.out.println("收到回复" + response2.getStatusLine());
            }

            @Override
            public void failed(final Exception ex) {
                latch.countDown();
                System.out.println("发生异常" + ":" + ex.getMessage());
            }

            @Override
            public void cancelled() {
                latch.countDown();
                System.out.println("cancelled");
            }
        });
    // 因为httpAsyncClient不阻塞，所以需要在这里等待执行完再调用httpAsyncClient.close()
    // 不然还没收到回复就关了http连接
    // 实际使用中当然不需要这样每调用一次关闭一次，而是工程一次运行的整个生命周期都用同一个，停止运行时关闭一次
        latch.await();
        httpAsyncClient.close();
    }
}

3. 需要使用HttpAsyncClient的场景

  之前的这篇博客https://www.jianshu.com/p/4109f517e781提到，有个服务因为未知原因需要6分钟才能回复http请求。我们假设我们的系统是1000并发，6分钟我们需要发出去6*60*1000=360000请求。我们使用线程，每个线程每次发送一次请求，直到第七分钟的时候，第一个http请求才返回，这次请求的相关工作完成，线程才能释放这次请求，去执行下一个请求，那么我们就需要360000个线程的线程池，这也太恐怖了。
  这时候我们改用HttpAsyncClient，发送请求之后，这次请求任务就完成了，就收http返回交给HttpAsyncClient内部的一个机制（这个后面讲），这时候这个线程就得到了释放，可以去执行下一个请求。假设从处理数据准备发送http请求到发出去用了1秒钟，那么我们也只需要1000个核心线程的线程池就可以了。这是多么巨大的资源节省。（众所周知，java线程池还是很占用内存的，并且CPU在不同线程之间切换也很浪费时间）。
  我们来做一个模拟：
首先写一个接收http请求的服务：

@GetMapping("/get")
public void test() {
    try {
    // 延时五秒钟
        Thread.sleep(1000 * 5);
    }catch (Exception e) {
        e.printStackTrace();
    }
}

同步http请求

// 同步http工具类
import org.apache.http.client.config.RequestConfig;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;

public class HttpClientUtil {
    public static CloseableHttpClient getHttpClient() {
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(15000)
                .setConnectTimeout(5000)
                .setConnectionRequestTimeout(50000000)
                .build();
        Registry<ConnectionSocketFactory> registry = RegistryBuilder.<ConnectionSocketFactory>create()
                .register("HTTP", new PlainConnectionSocketFactory()).build();
        PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager(registry);
    // 最大http连接
        connectionManager.setMaxTotal(1000);
        // 每个host的最大连接
        connectionManager.setDefaultMaxPerRoute(1000);
        connectionManager.setValidateAfterInactivity(500);
        return HttpClients.custom()
                .setDefaultRequestConfig(requestConfig)
                .setConnectionManager(connectionManager)
                .setConnectionManagerShared(true)
                .build();
    }
}

// 发送同步http请求
// 线程池只有100个线程，发送1000个请求，每个线程要用10次
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestHttpClient {
    public static void main(String[] args) throws Exception {
        System.out.println(System.currentTimeMillis());
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 100, 2, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10000), new ThreadPoolExecutor.AbortPolicy());
        CloseableHttpClient httpClient = HttpClientUtil.getHttpClient();
        CountDownLatch latch = new CountDownLatch(1000);
        for (int i = 0; i < 1000; i++) {
            threadPoolExecutor.execute(new MyRunnable(httpClient, i, latch));
        }
        latch.await();
        httpClient.close();
        threadPoolExecutor.shutdown();
        System.out.println(System.currentTimeMillis());
    }

    private static class MyRunnable implements Runnable {
        private CloseableHttpClient httpclient;
        private int i;
        private CountDownLatch latch;

        public MyRunnable(CloseableHttpClient httpclient, int i, CountDownLatch latch) {
            this.httpclient = httpclient;
            this.i = i;
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                final HttpGet request = new HttpGet("http://localhost:8043/get");
                System.out.println("发送请求" + i);
                HttpResponse response = httpclient.execute(request);
                System.out.println(response.getStatusLine());
                System.out.println("发送请求" + i + "完成");
                latch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

// 异步HttpAsyncClient工具类
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;

public class HttpAsyncClientUtil {

    public static CloseableHttpAsyncClient getHttpAsyncClient() {
        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectTimeout(50000)
                .setSocketTimeout(50000)
                .setConnectionRequestTimeout(1000000000)
                .build();
        //配置io线程
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom().
                setIoThreadCount(10)
                .setSoKeepAlive(true)
                .build();
        //设置连接池大小
        ConnectingIOReactor ioReactor=null;
        try {
            ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
        } catch (IOReactorException e) {
            e.printStackTrace();
        }
        PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
        connManager.setMaxTotal(1000);
        connManager.setDefaultMaxPerRoute(1000);
        return HttpAsyncClients.custom().
                setConnectionManager(connManager)
                .setDefaultRequestConfig(requestConfig)
                .build();
    }
}

// 发异步http请求
// 线程池只有100个线程，1000个请求，每个线程需要用10次
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestAsyncHttpClient {

    public static void main(String[] args) throws Exception {
        Thread.sleep(20000);
        System.out.println(System.currentTimeMillis());
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 100, 2, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.AbortPolicy());
        CloseableHttpAsyncClient httpAsyncClient = HttpAsyncClientUtil.getHttpAsyncClient();
        CountDownLatch latch = new CountDownLatch(1000);
        List<CallbackTask> respList = new LinkedList<>();
        httpAsyncClient.start();
        for (int i = 0; i < 1000; i++) {
            CallbackTask callbackTask = new CallbackTask(latch, Integer.toString(i));
            respList.add(callbackTask);
            threadPoolExecutor.execute(new MyRunnable(httpAsyncClient, i, latch, callbackTask));
        }
        System.out.println("发送请求完成");
        latch.await();
        httpAsyncClient.close();
        threadPoolExecutor.shutdown();
        System.out.println(System.currentTimeMillis());
    }

    private static class MyRunnable implements Runnable {
        private CloseableHttpAsyncClient httpclient;
        private int i;
        private CountDownLatch latch;
        private CallbackTask callbackTask;
        public MyRunnable(CloseableHttpAsyncClient httpclient, int i, CountDownLatch latch, CallbackTask callbackTask) {
            this.httpclient = httpclient;
            this.i = i;
            this.latch = latch;
            this.callbackTask = callbackTask;
        }

        @Override
        public void run() {
            try {
                final HttpGet request = new HttpGet("http://localhost:8043/get");
                System.out.println("发送请求" + i);
                httpclient.execute(request, callbackTask);
                System.out.println("发送请求" + i + "完成");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

// 用到的等待处理回调的类
import org.apache.http.HttpResponse;
import org.apache.http.concurrent.FutureCallback;

import java.util.concurrent.CountDownLatch;

public class CallbackTask implements FutureCallback<HttpResponse> {
    private CountDownLatch latch;

    // 区分http请求
    private String name;

    public CallbackTask(CountDownLatch latch, String name) {
        this.latch = latch;
        this.name = name;
    }

    @Override
    public void completed(final HttpResponse response2) {
        latch.countDown();
        System.out.println("收到回复" + name + response2.getStatusLine());
    }

    @Override
    public void failed(final Exception ex) {
        latch.countDown();
        System.out.println(name + "发生异常" + ":" + ex.getMessage());
    }

    @Override
    public void cancelled() {
        latch.countDown();
        System.out.println(name + " cancelled");
    }
}

  上面的例子，同步请求用时50s多，异步请求用时25s多。说明同步请求每次发送100个请求，接收到http返回，线程释放资源，才发送下一次100个请求。一共5*10=50s。而异步请求直接发送出去1000个请求，一起等待后的返回，至于为什么是25秒而不是5秒，因为线程切换需要时间。
  请求数越大，这两个时间差距越明显（我试了2000个请求只用了50秒）。但是需要保证http最大连接数>=一次性发出去所有请求数。

4. 这里还有一个问题

  HttpAsyncClient内部的等待回调是如何实现的，如果还是每一个等待开一个线程，那资源消耗还是很大，得不偿失。
  这当然是不可能的。
  我们现在在运行过程中用jconsole看一下。除了内存和CPU多占了（因为同样的时间里处理了更多的任务，所以可以接受），只多了几个IO线程，并没有多占额外资源。

image.png
image.png