使用CountdownLatch和线程池批量处理http请求，并处理响应数据

背景和问题

​ 背景：最近项目的一个接口数据，需要去请求其他多个服务器的数据，然后统一返回；
问题点：如果遍历所有的服务器地址，然后串行请求就会出现请求时间过长，加入需要请求十个服务器，一个服务器是1s那么请求服务器数据总时间就需要10s，导致响应时间太长，所以需要使用多线程。如果直接使用多线程去请求，那么没法知道是否所有接口是否都请求结束，所以用到了技术门闩CountdownLatch，每一个接口请求结束之后都会调用CountdownLatch的count方法进行计数，当归零后就会唤醒主线程进行后续逻辑，并且使用ConcurrentLinkedQueue记录响应结果。

话不多说，直接上代码！！

准备数据：

​ 四个接口（三个模拟请求服务器接口，一个直接访问的接口），由于我是本地环境，所以在每个接口中设置了不同的休眠时间，来模拟不同服务器场景

代码展示

1. 模拟接口

   	@RequestMapping("/hello")public String hello(@RequestParam(value = "id") Long id, @RequestBody User params) {if (id != 1) {return null;}System.out.println(params.toString());try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}List<User> users = new ArrayList<>();users.add(new User("张三", 1, "男"));users.add(new User("李四", 2, "男"));users.add(new User("王五", 3, "女"));return JSON.toJSONString(users);}@RequestMapping("/hello1")public String hello1(@RequestParam(value = "id") Long id) {if (id != 2) {return null;}try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {throw new RuntimeException(e);}List<User> users = new ArrayList<>();users.add(new User("张三1", 11, "男"));users.add(new User("李四1", 21, "男"));users.add(new User("王五1", 31, "女"));return JSON.toJSONString(users);}@RequestMapping("/hello2")public String hello2(@RequestParam(value = "id") Long id) {if (id != 3) {return null;}try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {throw new RuntimeException(e);}List<User> users = new ArrayList<>();users.add(new User("张三2", 12, "男"));users.add(new User("李四2", 22, "男"));users.add(new User("王五2", 32, "女"));return JSON.toJSONString(users);}	
   

2. 直接访问接口数据

    @RequestMapping("/demo")public String demo() throws InterruptedException, IOException {OkHttpClient client = new OkHttpClient();final CountDownLatch countDownLatch = new CountDownLatch(3);// 使用ConcurrentLinkedQueue 存储每个请求的响应结果，ConcurrentLinkedQueue 是一个线程安全的final ConcurrentLinkedQueue<Response> responses = new ConcurrentLinkedQueue<>();ExecutorService executor = Executors.newFixedThreadPool(10);long start = System.currentTimeMillis();for (int i = 1; i <= urls.size(); i++) {String url = urls.get(i - 1);int finalI = i;executor.submit(() -> {//构建请求中需要的请求参数 id 通过 RequestParam获取HttpUrl.Builder urlBuilder = HttpUrl.parse(url).newBuilder();urlBuilder.addQueryParameter("id", String.valueOf(finalI));String newUrl = urlBuilder.build().toString();// 表单提交
   //                FormBody formBody = new FormBody.Builder().add("id", String.valueOf(finalI)).build();
   //                Request request = new Request.Builder().url(newUrl).post(formBody).build();// 构建参数，通过@RequestBody取出参数User user = new User("1", 2, "男");okhttp3.RequestBody requestBody = okhttp3.RequestBody.create(MediaType.parse("application/json; charset=utf-8"),JSON.toJSONString(user));Request request = new Request.Builder().url(newUrl).post(requestBody).build();try {Response response = client.newCall(request).execute();if (!response.isSuccessful()) {// 可以在单独记录下，然后做异常处理throw new IOException("请求失败：" + response);} else {responses.add(response);}} catch (IOException e) {throw new RuntimeException(e);} finally {// 执行一个请求进行一次计数countDownLatch.countDown();}});}//等待countDownlatch 计数门闩归零即所有线程请求完成，然后唤醒线程，但是会设置一个最长等待时长 10sboolean await = countDownLatch.await(10, TimeUnit.SECONDS);executor.shutdown();long end = System.currentTimeMillis();System.out.println("http的整个请求时间为：" + DateUtil.formatBetween(end - start));Map<String,  List<User>> res = new HashMap<>();if (!responses.isEmpty()) {int i = 0;for (Response response : responses) {URL url = response.request().url().url();String string = response.body().string();List<User> users = JSON.parseArray(string, User.class);res.put(url.toString(),users);}} else {System.out.println("无响应结果！");}System.out.println(res);return "demo";}
   

3. 其他相关信息

     private static final List<String> urls = new ArrayList<>();static {urls.add("http://localhost:8080/hello");urls.add("http://localhost:8080/hello1");urls.add("http://localhost:8080/hello2");}public static class User {private String name;private Integer age;private String sex;public User(String name, Integer age, String sex) {this.name = name;this.age = age;this.sex = sex;}public String getName() {return name;}public void setName(String name) {this.name = name;}public Integer getAge() {return age;}public void setAge(Integer age) {this.age = age;}public String getSex() {return sex;}public void setSex(String sex) {this.sex = sex;}@Overridepublic String toString() {return "User{" +"name='" + name + '\'' +", age=" + age +", sex='" + sex + '\'' +'}';}
   

           //相关依赖<dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.21</version></dependency><!-- https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp --><dependency><groupId>com.squareup.okhttp3</groupId><artifactId>okhttp</artifactId><version>3.5.0</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.38</version></dependency>
   

结果

结果解释

1. 第一行：是/hello接口中@RequestBody的参数打印。
2. 第二行是整个请求时间，因为我设置的/hello2接口时间为5s，可以看到这里的接口请求时间是最长的接口时间，而不是所有时间相加。
3. 可以看到设置的参数能够成功获取，并且数据能成功接收。

注意事项

1. response中body体的数据只能取一次，取出之后就会将其设置为closed，所以建议使用变量进行接收之后再做处理。

2. 这个唤醒时间最好设置一个默认值，免得程序出问题主线程一直卡死在这里。

   countDownLatch.await(10, TimeUnit.SECONDS);
   

3. 这个count一定不能忘了，不然这个主线程也就卡死了

最后，这只是一个简单的demo程序，真实的项目情况肯定是不一样的，所以只能做一个参考，具体情况还需做具体处理！

源码已提交gitee