一、背景

现实业务开发中，通常为了避免超时、对方接口限制等原因需要对支持批量的接口的数据分批调用。

比如List参数的size可能为几十个甚至上百个，但是假如对方dubbo接口比较慢，传入50个以上会超时，那么可以每次传入20个，分批执行。

通常很多人会写 for 循环或者 while 循环，非常不优雅，无法复用，而且容易出错。

下面结合 Java8 的 Stream ，Function ,Consumer 等特性实现分批调用的工具类封装和自测。

并给出 CompletableFuture 的异步改进方案。

二、实现

工具类：

package com.chujianyun.common.java8.function;import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;/*** 执行工具类** @author 明明如月*/
public class ExecuteUtil {public static <T> void partitionRun(List<T> dataList, int size, Consumer<List<T>> consumer) {if (CollectionUtils.isEmpty(dataList)) {return;}Preconditions.checkArgument(size > 0, "size must not be a minus");Lists.partition(dataList, size).forEach(consumer);}public static <T, V> List<V> partitionCall2List(List<T> dataList, int size, Function<List<T>, List<V>> function) {if (CollectionUtils.isEmpty(dataList)) {return new ArrayList<>(0);}Preconditions.checkArgument(size > 0, "size must not be a minus");return Lists.partition(dataList, size).stream().map(function).filter(Objects::nonNull).reduce(new ArrayList<>(),(resultList1, resultList2) -> {resultList1.addAll(resultList2);return resultList1;});}public static <T, V> Map<T, V> partitionCall2Map(List<T> dataList, int size, Function<List<T>, Map<T, V>> function) {if (CollectionUtils.isEmpty(dataList)) {return new HashMap<>(0);}Preconditions.checkArgument(size > 0, "size must not be a minus");return Lists.partition(dataList, size).stream().map(function).filter(Objects::nonNull).reduce(new HashMap<>(),(resultMap1, resultMap2) -> {resultMap1.putAll(resultMap2);return resultMap1;});}
}

待调用的服务（模拟）

package com.chujianyun.common.java8.function;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;public class SomeManager {public void aRun(Long id, List<String> data) {}public List<Integer> aListMethod(Long id, List<String> data) {return new ArrayList<>(0);}public Map<String, Integer> aMapMethod(Long id, List<String> data) {return new HashMap<>(0);}
}

单元测试：

package com.chujianyun.common.java8.function;import org.apache.commons.lang3.RandomUtils;
import org.jeasy.random.EasyRandom;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.internal.verification.Times;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.modules.junit4.PowerMockRunner;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;@RunWith(PowerMockRunner.class)
public class ExecuteUtilTest {private EasyRandom easyRandom = new EasyRandom();@Mockprivate SomeManager someManager;// 测试数据private List<String> mockDataList;private int total = 30;@Beforepublic void init() {// 构造30条数据mockDataList = easyRandom.objects(String.class, 30).collect(Collectors.toList());}@Testpublic void test_a_run_partition() {// mock aRunPowerMockito.doNothing().when(someManager).aRun(anyLong(), any());// 每批 10 个ExecuteUtil.partitionRun(mockDataList, 10, (eachList) -> someManager.aRun(1L, eachList));//验证执行了 3 次Mockito.verify(someManager, new Times(3)).aRun(anyLong(), any());}@Testpublic void test_call_return_list_partition() {// mock  每次调用返回条数(注意每次调用都是这2个)int eachReturnSize = 2;PowerMockito.doReturn(easyRandom.objects(String.class, eachReturnSize).collect(Collectors.toList())).when(someManager).aListMethod(anyLong(), any());// 分批执行int size = 4;List<Integer> resultList = ExecuteUtil.partitionCall2List(mockDataList, size, (eachList) -> someManager.aListMethod(2L, eachList));//验证执行次数int invocations = 8;Mockito.verify(someManager, new Times(invocations)).aListMethod(anyLong(), any());// 正好几轮int turns;if (total % size == 0) {turns = total / size;} else {turns = total / size + 1;}Assert.assertEquals(turns * eachReturnSize, resultList.size());}@Testpublic void test_call_return_map_partition() {// mock  每次调用返回条数// 注意：// 如果仅调用doReturn一次，那么每次返回都是key相同的Map，// 如果需要不覆盖，则doReturn次数和 invocations 相同）int eachReturnSize = 3;PowerMockito.doReturn(mockMap(eachReturnSize)).doReturn(mockMap(eachReturnSize)).when(someManager).aMapMethod(anyLong(), any());// 每批int size = 16;Map<String, Integer> resultMap = ExecuteUtil.partitionCall2Map(mockDataList, size, (eachList) -> someManager.aMapMethod(2L, eachList));//验证执行次数int invocations = 2;Mockito.verify(someManager, new Times(invocations)).aMapMethod(anyLong(), any());// 正好几轮int turns;if (total % size == 0) {turns = total / size;} else {turns = total / size + 1;}Assert.assertEquals(turns * eachReturnSize, resultMap.size());}private Map<String, Integer> mockMap(int size) {Map<String, Integer> result = new HashMap<>(size);for (int i = 0; i < size; i++) {// 极力保证key不重复result.put(easyRandom.nextObject(String.class) + RandomUtils.nextInt(), easyRandom.nextInt());}return result;}}

注意：

1 判空

.filter(Objects::nonNull)

这里非常重要，避免又一次调用返回 null，而导致空指针异常。

2 实际使用时可以结合apollo配置，灵活设置每批执行的数量，如果超时随时调整

3 用到的类库

集合工具类： commons-collections4、guava （可以不用）

这里的list划分子list也可以使用stream的 skip ,limit特性自己去做，集合判空也可以不借助collectionutils.

构造数据：easy-random

单元测试框架： Junit4 、 powermockito、mockito

4 大家可以加一些更强大的功能，如允许设置每次调用的时间间隔、并行或并发调用等。

三、改进

以上面的List接口为例，将其改为异步版本：

    public static <T, V> List<V> partitionCall2ListAsync(List<T> dataList,int size,ExecutorService executorService,Function<List<T>, List<V>> function) {if (CollectionUtils.isEmpty(dataList)) {return new ArrayList<>(0);}Preconditions.checkArgument(size > 0, "size must not be a minus");List<CompletableFuture<List<V>>> completableFutures = Lists.partition(dataList, size).stream().map(eachList -> {if (executorService == null) {return CompletableFuture.supplyAsync(() -> function.apply(eachList));} else {return CompletableFuture.supplyAsync(() -> function.apply(eachList), executorService);}}).collect(Collectors.toList());CompletableFuture<Void> allFinished = CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]));try {allFinished.get();} catch (Exception e) {throw new RuntimeException(e);}return completableFutures.stream().map(CompletableFuture::join).filter(CollectionUtils::isNotEmpty).reduce(new ArrayList<V>(), ((list1, list2) -> {List<V> resultList = new ArrayList<>();if(CollectionUtils.isNotEmpty(list1)){resultList.addAll(list1);}if(CollectionUtils.isNotEmpty(list2)){resultList.addAll(list2);}return resultList;}));}

测试代码：

    // 测试数据private List<String> mockDataList;private int total = 300;private AtomicInteger atomicInteger;@Beforepublic void init() {// 构造total条数据mockDataList = easyRandom.objects(String.class, total).collect(Collectors.toList());}@Testpublic void test_call_return_list_partition_async() {ExecutorService executorService = Executors.newFixedThreadPool(10);atomicInteger = new AtomicInteger(0);Stopwatch stopwatch = Stopwatch.createStarted();// 分批执行int size = 2;List<Integer> resultList = ExecuteUtil.partitionCall2ListAsync(mockDataList, size, executorService, (eachList) -> someCall(2L, eachList));Stopwatch stop = stopwatch.stop();log.info("执行时间: {} 秒", stop.elapsed(TimeUnit.SECONDS));Assert.assertEquals(total, resultList.size());// 正好几轮int turns;if (total % size == 0) {turns = total / size;} else {turns = total / size + 1;}log.info("共调用了{}次", turns);Assert.assertEquals(turns, atomicInteger.get());// 顺序也一致for(int i =0; i< mockDataList.size();i++){Assert.assertEquals((Integer) mockDataList.get(i).length(), resultList.get(i));}}/*** 模拟一次调用*/private List<Integer> someCall(Long id, List<String> strList) {log.info("当前-->{}，strList.size：{}", atomicInteger.incrementAndGet(), strList.size());try {TimeUnit.SECONDS.sleep(2L);} catch (InterruptedException e) {e.printStackTrace();}return strList.stream().map(String::length).collect(Collectors.toList());}

通过异步可以尽可能快得拿到执行结果。

四、总结

1 要灵活运用Java 8 的特性简化代码

2 要注意代码的封装来使代码更加优雅,复用性更强

3 要利用来构造单元测试的数据框架如 java-faker和easy-random来提高构造数据的效率

4 要了解性能改进的常见思路：合并请求、并发、并行、缓存等。

--------------------------------------------

我在参见 CSDN 1024 程序员活动（2019年10月24日截止）

如果我的博客对你有帮助，且有时间，欢迎浏览器后者微信扫码，帮我点赞支持我：

Java 数据分批调用接口的正确姿势相关推荐

java多线程分批调用接口
线程池工具类 public class ThreadPoolUtil {public static final long DEFAULT_WAIT_SECONDS = 5000;private sta ...
python调用第三方接口获取数据_python调用接口,python接收post请求接口(附完整代码)...
与Scala语言相比,Python有其独特的优势和广泛的应用,python调用接口,因此Spark也推出了PySpark,它在框架上提供了一个使用Python语言的接口,python接收post请求接 ...
antd pro中如何使用mock数据以及调用接口
antd pro的底层基础框架使用的是dva,dva采用effect的方式来管理同步化异步在dva中主要分为3层 services models components models层用于存放数据 ...
Java线程池「异常处理」正确姿势：有病就得治
假设我们有一个线程池,由于程序需要,我们向该线程池中提交了好多好多任务,但是这些任务都没有对异常进行try catch处理,并且运行的时候都抛出了异常 .这会对线程池的运行带来什么影响? 正确答案是 ...
Java枚举(Enum)类型使用的正确姿势
关于Java Enum的介绍及原理参见 Java枚举(Enum)类型的基本介绍与原理探求 Enum类型的基本使用定义一个枚举类的主要作用就是在逻辑代码中对某个变量的值进行比较.同样以季节的枚举类Se ...
VS 调用 scanf 的正确姿势
欢迎来到 Claffic 的博客
老黄历java_基于聚合数据的老黄历接口调用示例-JAVA版
本文介绍聚合数据的老黄历接口的使用依赖 net.sf.json-lib json-lib 2.2.3 jdk15 代码部分 package com.example.demo; import net. ...
获取java返回的数据_java调用第三方接口，获取接口返回的数据。
java接收远程调用的数据,得到的是如上个数的返回内容,我怎么写才能获取到值,现在使用的请求方法如下: public static HttpResult postJsonData(String url ...
Java调用接口获取json数据解析后保存到数据库
文章目录 1.在yml文件中配置自己定义的接口URL 2.在Controller中添加请求方法和路径 3.在Service中添加方法 4.在ServiceImpl中实现方法今天给大家带来一个调用接口 ...

Java 数据分批调用接口的正确姿势

一、背景

二、实现

三、改进

四、总结

Java 数据分批调用接口的正确姿势相关推荐

最新文章

热门文章