快捷導(dǎo)航

java如何多線程批量更新10萬級的數(shù)據(jù)

更新時間：2024年10月12日 08:42:30 作者：carter171717

在處理大數(shù)據(jù)量的批量更新時,直接使用mybatis的updateBatch可能導(dǎo)致效率低下甚至OOM,通過每次處理5000條數(shù)據(jù)的方式雖然安全但效率低,更優(yōu)的解決方案是使用多線程處理,將數(shù)據(jù)分批并多線程執(zhí)行,有效提高了處理速度并保證了系統(tǒng)穩(wěn)定性

java多線程批量更新10萬級的數(shù)據(jù)

好久沒有寫文章，今天剛好沒啥事，就動手記錄一下，好記性不如爛筆頭！言歸正傳，我最近接到的一個工作任務(wù)大概內(nèi)容是，有一張數(shù)據(jù)量在十萬+級別的表，需要新增一個字段，并且要寫入初始化值。

業(yè)務(wù)其實(shí)非常的簡單，全部查詢出來一個列表，然后用mybatis的updateBatch批量更新，其實(shí)在我的實(shí)踐過程中也沒什么問題，但是執(zhí)行的效率是很低的，而且一旦數(shù)據(jù)量過大，如果機(jī)器配置不太行的話，很可能會直接OOM，如果在正式環(huán)境出現(xiàn)這個問題，那完?duì)僮?，?zhǔn)備刪庫跑路！

所以呢，我就想了一個比較保險但是比較低級的辦法，每次查詢出5000條數(shù)據(jù)，去做批量更新，確保內(nèi)存不會溢出導(dǎo)致服務(wù)崩盤，這當(dāng)然也是可以解決問題，但是就是修復(fù)數(shù)據(jù)需要執(zhí)行很多次，顯得比較愚蠢一點(diǎn)。

如何用比較方便并高效的方式來修復(fù)大數(shù)量的數(shù)據(jù)

第一反應(yīng)肯定是多線程啦，方案是：

1.查詢出全部的數(shù)據(jù)（10萬條）
2.對數(shù)據(jù)進(jìn)行分批，每批5000條，
3.多線程同時處理多批數(shù)據(jù)
4.等待執(zhí)行完成，返回成功

直接上核心代碼

寫一個通用的分批工具類，把一個List集合，拆分成多個小的List集合

/**
 * 拆分集合
 *
 * @param <T> 泛型對象
 * @param resList 需要拆分的集合
 * @param subListLength 每個子集合的元素個數(shù)
 * @return 返回拆分后的各個集合組成的列表
 **/
public static <T> List<List<T>> splitList(List<T> resList, int subListLength) {
    if (CollectionUtils.isEmpty(resList) || subListLength <= 0) {
        return new ArrayList<>();
    }
    List<List<T>> ret = new ArrayList<>();
    int size = resList.size();
    if (size <= subListLength) {
        // 數(shù)據(jù)量不足 subListLength 指定的大小
        ret.add(resList);
    } else {
        int pre = size / subListLength;
        int last = size % subListLength;
        // 前面pre個集合，每個大小都是 subListLength 個元素
        for (int i = 0; i < pre; i++) {
            List<T> itemList = new ArrayList<>(subListLength);
            for (int j = 0; j < subListLength; j++) {
                itemList.add(resList.get(i * subListLength + j));
            }
            ret.add(itemList);
        }

        // last的進(jìn)行處理
        if (last > 0) {
            List<T> itemList = new ArrayList<>(last);
            for (int i = 0; i < last; i++) {
                itemList.add(resList.get(pre * subListLength + i));
            }
            ret.add(itemList);
        }
    }
    return ret;
}

然后就是用多線程業(yè)務(wù)處理了，代碼如下

 public static void doThreadBusiness(List<String> totalList) {
        Long startTime = System.currentTimeMillis();
        System.out.println("本次更新任務(wù)開始");
        System.out.println("本機(jī)CPU核心數(shù)："+Runtime.getRuntime().availableProcessors());
        List<String> updateList = new ArrayList();
        // 初始化線程池, 參數(shù)一定要一定要一定要調(diào)好?。。?！
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 50,
                4, TimeUnit.SECONDS, new ArrayBlockingQueue(10), new ThreadPoolExecutor.DiscardPolicy());
        // 大集合拆分成N個小集合，然后用多線程去處理數(shù)據(jù)，確保不會因?yàn)閿?shù)據(jù)量過大導(dǎo)致執(zhí)行過慢
        List<List<String>> splitNList = SplitListUtils.splitList(totalList, 5000);
        // 記錄單個任務(wù)的執(zhí)行次數(shù)
        CountDownLatch countDownLatch = new CountDownLatch(splitNList.size());
        // 對拆分的集合進(jìn)行批量處理, 先拆分的集合, 再多線程執(zhí)行
        for (List<String> singleList : splitNList) {
            // 線程池執(zhí)行
            threadPool.execute(new Thread(new Runnable(){
                @Override
                public void run() {
                    //模擬執(zhí)行時間
                    System.out.println("當(dāng)前線程："+Thread.currentThread().getName());
                    List<String> batchUpdateVipList = new ArrayList<>();
                    for (String str : singleList) {
                        //組裝要執(zhí)行的批量更新數(shù)據(jù)
                        batchUpdateVipList.add(str);
                    }
                    // 這里模擬執(zhí)行數(shù)據(jù)庫批量更新操作
                    System.out.println("本次批量更新數(shù)據(jù)量："+ batchUpdateVipList.size());
                    // 任務(wù)個數(shù) - 1, 直至為0時喚醒a(bǔ)wait()
                    countDownLatch.countDown();
                }
            }));
        }
        try {
            // 讓當(dāng)前線程處于阻塞狀態(tài)，直到鎖存器計(jì)數(shù)為零
            countDownLatch.await();
        } catch (Exception e) {
            System.out.println("系統(tǒng)出現(xiàn)異常");
        }
        Long endTime = System.currentTimeMillis();
        Long useTime = endTime - startTime;
        System.out.println("本次更新任務(wù)結(jié)束,共計(jì)用時"+useTime+"毫秒");
    }

代碼很簡單一看就懂了，這里要說一下CountDownLatch的使用，其實(shí)開發(fā)中并不常用，但是面試卻很常用，這里蠻寫一下，我使用CountDownLatch來阻塞主線程，等待多線程執(zhí)行完畢后，再繼續(xù)主線程，返回更新的結(jié)果，這個場景其實(shí)很經(jīng)常使用到。

如果不用CountDownLatch，主線程會馬上返回，如果是數(shù)據(jù)量大的情況下，往往會執(zhí)行蠻久的，但是結(jié)果秒返回，就會給人一種錯覺。

CountDownLatch countDownLatch = new CountDownLatch(splitNList.size());

在線程執(zhí)行完畢后，需要調(diào)用一下countDown，

// 任務(wù)個數(shù) - 1, 直至為0時喚醒a(bǔ)wait()
countDownLatch.countDown();

在主線程用await（）進(jìn)行阻塞等待，這樣主線程就會一直等到所有的子線程都執(zhí)行完成了，繼續(xù)執(zhí)行主線程的后續(xù)代碼

 countDownLatch.await();

其實(shí)這里面還有一個非常重要的面試點(diǎn)，就是多線程的七大參數(shù)如何設(shè)置，

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 
50,
4, 
TimeUnit.SECONDS, 
new ArrayBlockingQueue(10), 
new ThreadPoolExecutor.DiscardPolicy()
);

有興趣了解具體是設(shè)置方法可以另行查詢資料，這也是面試必問的考點(diǎn)。

最后貼一下返回的打印結(jié)果吧，如下圖所示：