Java多线程处理千万级数据更新操作_Java

故事背景

之前因为用户信息安全把好几个敏感信息用aes加密保存了，应业务需求，需要通过后台userid等信息查询订单，所以需要明文保存数据。但是目前订单表已经有7000万+订单数据，操作起来比较麻烦，肯定需要分批操作的，否则会有内存溢出等问题。初步想到的解决方案是新建一个数据库表，保存处理订单表的id是处理到哪一条数据，然后用limit查询一万条数据，处理完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(){
处理数据
update数据
}
id = list.get(10000).getid();
updateid();
}

这样update数据处理可能会比较慢，可以用多线程处理。

方案一

用多线程的话需要考虑一下怎么设计，首先是想着拿到一万条数据之后，分成10份，给10条线程去处理数据，但是这样可能会因为线程处理太慢，线程任务队列太多导致oom异常，所以可以等待线程全都执行完毕再处理下一批数据。那么怎么等待线程执行完再处理呢？百度完发现可以用countdownlatch实现~

方案二

如果不想等线程执行完再处理，可以转换下思路。把list当做消息队列，每个线程获取1千条数据，如果队列数据为空则再查询后面1万条数据做处理。这里有一个多线程处理共享队列的问题，所以需要在取数据的时候和增删队列数据的时候加锁。

伪代码

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){
        ...
        }
    }
}

多线程的坑

多线程一定要注意数据的并发安全问题，比如说用sublist的话并不会复制一个新的list出来，只是原有list的基础上的一个视图，如果原来的list改变，sublist得到的list也会跟着改变，如果想要用sublist一定要new一个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条数据
        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()+"开始打印数据======");
        for(int i=0;i<list.size();i++){

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

}

如果sleep5秒模拟队列的增删操作，原来的list已经变了，会影响线程里的list变动，会报错

java.util.concurrentmodificationexception

要解决这个问题需要用创建一个新的list，这就不会受原来的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()) ;

后面还遇到一个问题，new出来的多线程类无法使用@autowired注入，解决办法：

1、将需要的bean作为线程的的构造函数的参数传入

2、使用applicationcontext.getbean方法来静态的获取bean（推荐）