快捷導(dǎo)航

Java使用線程池批量處理數(shù)據(jù)操作具體流程

更新時間：2023年06月26日 10:06:49 作者：咕嚕咕嚕虎

這篇文章主要給大家介紹了關(guān)于Java使用線程池批量處理數(shù)據(jù)操作的相關(guān)資料,Java多線程編程中線程池是一個非常重要的概念,線程池可以提高線程的復(fù)用率和任務(wù)調(diào)度的效率,尤其是當(dāng)需要查詢大批量數(shù)據(jù)時,需要的朋友可以參考下

疑問&思路：

1.如何保證數(shù)據(jù)按順序批量處理

2.如何保證數(shù)據(jù)全部處理完統(tǒng)一返回

3.如何保證是多任務(wù)異步操作

4.如何提高運行效率，減少運行時間

1.使用ArrayList 插入數(shù)據(jù)有序且可重復(fù)

2.CountDownLatch / Future / CompletableFuture

3.多線程

4.線程池創(chuàng)建多線程

具體流程：

獲取需要進行批量更新的大集合oldList，對大集合進行拆分操作，分成N個小集合nweList-1 ~ nweList-N 。
開啟線程池，針對集合的大小進行調(diào)參，對小集合進行批量更新操作。
對流程進行控制，控制線程執(zhí)行順序。

創(chuàng)建List分割工具類：

public class ListSplitUtils {
    //這里使用泛型T 接收  做到通用工具類
    //resList總數(shù)據(jù)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) {  //指定數(shù)據(jù)過小直接處理
            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;
    }

創(chuàng)建線程池：

// 初始化線程池
/**
* corePoolSize： 一直保持的線程的數(shù)量，即使線程空閑也不會釋放。除非設(shè)置了 allowCoreThreadTimeout 為 true；
* maxPoolSize：允許最大的線程數(shù)，隊列滿時開啟新線程直到等于該值；
* keepAliveTime：表示空閑線程的存活時間。當(dāng)線程空閑時間達到keepAliveTime，該線程會退出，直到線程數(shù)量等于corePoolSize。只有當(dāng)線程池中的線程數(shù)大于corePoolSize時keepAliveTime才會起作用，直到線程中的線程數(shù)不大于corepoolSIze；
* TimeUnitunit：表示keepAliveTime的單位；
* workQueue：緩存任務(wù)的隊列；
* handler：表示當(dāng) workQueue 已滿，且池中的線程數(shù)達到 maxPoolSize 時，線程池拒絕添加新任務(wù)時采取的策略。
*/
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(20, 50, 4, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.AbortPolicy());
        //大集合拆分成N個小集合，保證多線程異步執(zhí)行, 過大容易回到單線程
        List<List<CheckRecordDetailsDanger>> splitNList = ListSplitUtils.split(CheckRecordDetailsDangerPage, 100); //先設(shè)置100  100以內(nèi)不考慮性能
        // 記錄單個任務(wù)的執(zhí)行次數(shù)
        CountDownLatch countDownLatch = new CountDownLatch(splitNList.size());
        for (List<CheckRecordDetailsDanger> singleList : splitNList) {
            // 線程池執(zhí)行
            threadPool.execute(new Thread(() -> {
                for (CheckRecordDetailsDanger checkRecordDetailsDanger : singleList) {
                    //統(tǒng)一賦值方法
                    //unifySetData(checkRecordDetailsDanger);  這是我的方法，需要替換成自己的處理邏輯
                    countDownLatch.countDown();
                }
            }));
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }