java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > Java线程池批量处理数据

Java使用线程池批量处理数据操作具体流程

作者:咕噜咕噜虎

这篇文章主要给大家介绍了关于Java使用线程池批量处理数据操作的相关资料,Java多线程编程中线程池是一个非常重要的概念,线程池可以提高线程的复用率和任务调度的效率,尤其是当需要查询大批量数据时,需要的朋友可以参考下

疑问&思路:

1.如何保证数据按顺序批量处理

2.如何保证数据全部处理完统一返回

3.如何保证是多任务异步操作

4.如何提高运行效率,减少运行时间

1.使用ArrayList 插入数据有序且可重复

2.CountDownLatch / Future / CompletableFuture

3.多线程

4.线程池创建多线程

具体流程:

创建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 。

到此这篇关于Java使用线程池批量处理数据操作的文章就介绍到这了,更多相关Java线程池批量处理数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文