Java使用线程池批量处理数据操作
Java使用线程池批量处理数据操作
疑问&思路：
1.如何保证数据按顺序批量处理
2.如何保证数据全部处理完统一返回
3.如何保证是多任务异步操作
4.如何提高运行效率，减少运行时间
1.使用ArrayList 插入数据有序且可重复
2.CountDownLatch / Future / CompletableFuture
3.多线程
4.线程池创建多线程
具体流程：
- 获取需要进行批量更新的大集合oldList，对大集合进行拆分操作，分成N个小集合nweList-1 ~ nweList-N 。
- 开启线程池，针对集合的大小进行调参，对小集合进行批量更新操作。
- 对流程进行控制，控制线程执行顺序。
创建List分割工具类：
public class ListSplitUtils {//这里使用泛型T 接收 做到通用工具类//resList总数据List subListLength:需要切割的长度public static <T> List<List<T>> split(List<T> resList, int subListLength) { if (CollectionUtils.isEmpty(resList) || subListLength <= 0) {return Lists.newArrayList();}List<List<T>> ret = Lists.newArrayList();int size = resList.size();if (size <= subListLength) { //指定数据过小直接处理ret.add(resList);} else {int n = size / subListLength;int last = size % subListLength;// 分成n个集合，每个大小都是 subListLength 个元素for (int i = 0; i < n; i++) {List<T> itemList = Lists.newArrayList();for (int j = 0; j < subListLength; j++) {itemList.add(resList.get(i * subListLength + j));}ret.add(itemList);}// last的进行处理if (last > 0) {List<T> itemList = Lists.newArrayList();for (int i = 0; i < last; i++) {itemList.add(resList.get(n* subListLength + i));}ret.add(itemList);}}return ret;}
创建线程池：
// 初始化线程池
/**
* corePoolSize： 一直保持的线程的数量，即使线程空闲也不会释放。除非设置了 allowCoreThreadTimeout 为 true；
* maxPoolSize：允许最大的线程数，队列满时开启新线程直到等于该值；
* keepAliveTime：表示空闲线程的存活时间。当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。只有当线程池中的线程数大于corePoolSize时keepAliveTime才会起作用，直到线程中的线程数不大于corepoolSIze；
* TimeUnitunit：表示keepAliveTime的单位；
* workQueue：缓存任务的队列；
* handler：表示当 workQueue 已满，且池中的线程数达到 maxPoolSize 时，线程池拒绝添加新任务时采取的策略。
*/ThreadPoolExecutor threadPool = new ThreadPoolExecutor(20, 50, 4, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.AbortPolicy());//大集合拆分成N个小集合，保证多线程异步执行, 过大容易回到单线程List<List<CheckRecordDetailsDanger>> splitNList = ListSplitUtils.split(CheckRecordDetailsDangerPage, 100); //先设置100 100以内不考虑性能// 记录单个任务的执行次数CountDownLatch countDownLatch = new CountDownLatch(splitNList.size());for (List<CheckRecordDetailsDanger> singleList : splitNList) {// 线程池执行threadPool.execute(new Thread(() -> {for (CheckRecordDetailsDanger checkRecordDetailsDanger : singleList) {//统一赋值方法//unifySetData(checkRecordDetailsDanger); 这是我的方法，需要替换成自己的处理逻辑countDownLatch.countDown();}}));}try {countDownLatch.await();} catch (InterruptedException e) {e.printStackTrace();}
后话
学习过程中可以了解一下 CountDownLatch 和 Future 以及 ThreadPoolExecutor 。
本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！