快捷導(dǎo)航

Java多線程處理千萬級(jí)數(shù)據(jù)更新操作

更新時(shí)間：2024年11月22日 10:53:23 作者：瑪奇瑪丶

這篇文章主要為大家詳細(xì)介紹了Java如何通過多線程處理千萬級(jí)數(shù)據(jù)更新操作,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

故事背景

之前因?yàn)橛脩粜畔踩押脦讉€(gè)敏感信息用AES加密保存了，應(yīng)業(yè)務(wù)需求，需要通過后臺(tái)userId等信息查詢訂單，所以需要明文保存數(shù)據(jù)。但是目前訂單表已經(jīng)有7000萬+訂單數(shù)據(jù)，操作起來比較麻煩，肯定需要分批操作的，否則會(huì)有內(nèi)存溢出等問題。初步想到的解決方案是新建一個(gè)數(shù)據(jù)庫表，保存處理訂單表的id是處理到哪一條數(shù)據(jù)，然后用limit查詢一萬條數(shù)據(jù)，處理完id+10000。

偽代碼

//查詢id
while(id<maxid){
Integer id = mapper.selectList().get(0);
List list = mapper.selectList(Wrappers.<order>lambdaQuery()
        .gt(order::getId,id)
        .last("limit 10000");
for(){
處理數(shù)據(jù)
update數(shù)據(jù)
}
id = list.get(10000).getId();
updateId();
}

這樣update數(shù)據(jù)處理可能會(huì)比較慢，可以用多線程處理。

方案一

用多線程的話需要考慮一下怎么設(shè)計(jì)，首先是想著拿到一萬條數(shù)據(jù)之后，分成10份，給10條線程去處理數(shù)據(jù)，但是這樣可能會(huì)因?yàn)榫€程處理太慢，線程任務(wù)隊(duì)列太多導(dǎo)致OOM異常，所以可以等待線程全都執(zhí)行完畢再處理下一批數(shù)據(jù)。那么怎么等待線程執(zhí)行完再處理呢？百度完發(fā)現(xiàn)可以用CountDownLatch實(shí)現(xiàn)~

方案二

如果不想等線程執(zhí)行完再處理，可以轉(zhuǎn)換下思路。把list當(dāng)做消息隊(duì)列，每個(gè)線程獲取1千條數(shù)據(jù)，如果隊(duì)列數(shù)據(jù)為空則再查詢后面1萬條數(shù)據(jù)做處理。這里有一個(gè)多線程處理共享隊(duì)列的問題，所以需要在取數(shù)據(jù)的時(shí)候和增刪隊(duì)列數(shù)據(jù)的時(shí)候加鎖。

偽代碼

private static List list;
//線程處理
public void run(){
    List dataList;
    while(true){
        synchronized(xx.class){
            if(list.size() == 0){
                list = getList();
            }
            //再判斷一次，如果為空就是處理完了
            if(orderList == null || orderList.size() == 0) {
                return;
            }
            //取最后1千條
            dataList = new ArrayList<>(list.subList(list.size()-1000,list.size()));
            //減去最后一千條
            list = list.subList(0,list.size()-1000);
        }
        //處理dataList
        for(Order order:dataList){
        ...
        }
    }
}

多線程的坑

多線程一定要注意數(shù)據(jù)的并發(fā)安全問題，比如說用subList的話并不會(huì)復(fù)制一個(gè)新的list出來，只是原有l(wèi)ist的基礎(chǔ)上的一個(gè)視圖，如果原來的list改變，subList得到的list也會(huì)跟著改變，如果想要用subList一定要new一個(gè)List接收。

DEMO

public static void main(String[] args) {
    List<Integer> list = new ArrayList<>();
    for(int i=0;i<100;i++){
        list.add(i);
    }
    while(list.size() >=10){
        //取最后10條
        List<Integer> dataList = list.subList(list.size()-10,list.size());
        //List<Integer> dataList = list.stream().skip(Math.max(0, list.size() - 10)).collect(Collectors.toList()) ;
        ListThread listThread = new ListThread(dataList);
        listThread.start();
        //去掉最后10條數(shù)據(jù)
        list = list.subList(0,list.size()-10);
    }

    list.add(33333);
}


class ListThread extends Thread{
    private List<Integer> list;

    public ListThread(List list){
        this.list = list;
    }

    public void run(){
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(Thread.currentThread().getName()+"開始打印數(shù)據(jù)======");
        for(int i=0;i<list.size();i++){

            System.out.println(list.get(i));
        }
    }

}

如果sleep5秒模擬隊(duì)列的增刪操作，原來的list已經(jīng)變了，會(huì)影響線程里的list變動(dòng)，會(huì)報(bào)錯(cuò)

java.util.ConcurrentModificationException

要解決這個(gè)問題需要用創(chuàng)建一個(gè)新的list，這就不會(huì)受原來的list影響

List<Integer> dataList = new ArrayList<>(list.subList(list.size()-10,list.size()));
//或者使用stream流
List<Integer> dataList = list.stream().skip(Math.max(0, list.size() - 10)).collect(Collectors.toList()) ;

后面還遇到一個(gè)問題，new出來的多線程類無法使用@Autowired注入，解決辦法：

1、將需要的Bean作為線程的的構(gòu)造函數(shù)的參數(shù)傳入

2、使用ApplicationContext.getBean方法來靜態(tài)的獲取Bean（推薦）