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

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 213,616评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 91,020评论 3赞 387
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 159,078评论 0赞 349
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 57,040评论 1赞 285
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 66,154评论 6赞 385
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,265评论 1赞 292
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,298评论 3赞 412
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,072评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,491评论 1赞 306
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,795评论 2赞 328
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,970评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,654评论 4赞 337
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,272评论 3赞 318
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,985评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,223评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,815评论 2赞 365
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,852评论 2赞 351