多线程处理并有序整合数据方案

来源：意榕旅游网

方案设想

多线程异步“并行”处理待处理数据【for+线程池单例创实例和回收】
- 防止处理过程中线程数过大，内存溢出，导致处理失败，例如持续for中new Thread
保证并行的线程处理个数【CountDownLatch】
- 防止线程池未全部结束就开始进行处理，这里设置线程都返回出具再进行处理
异步处理线程安全key-Object数据结构【ConcurrentHashMap】
- 防止处理过程中漏处理，例如HashMap实测线程不安全，处理数据会<=实际数据量
业务耗时模拟【Thread.sleep】
- 模拟业务处理耗时，注意单独异常捕获，防止造成中断退出

样例

线程池

import com.peng.common.util.LogUtils;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class BusinessThreadPool {

    private ThreadPoolExecutor threadPool;
    private volatile static BusinessThreadPool instance = null;
    private int corePoolSize = 100;
    private int maximumPoolSize = 200;
    private long keepAliveTime = 500L;

    private BusinessThreadPool() {
        initPool();
    }

    public static BusinessThreadPool getInstance() {
        if (null == instance) {
            synchronized (BusinessThreadPool.class) {
                if (null == instance) {
                    instance = new BusinessThreadPool();
                }
            }
        }
        return instance;
    }

    private void initPool() {
        try {
            this.threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
                    keepAliveTime, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(maximumPoolSize),
                    new ThreadPoolExecutor.CallerRunsPolicy());

            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    threadPool.shutdown();
                }
            });
        } catch (Exception e) {
            LogUtils.root.error("Exception:{}", e);
        }
    }

    public void execute(Runnable thread) {
        this.threadPool.execute(thread);
    }

    public Future<?> submit(Runnable thread) {
        return this.threadPool.submit(thread);
    }

    public int getActivityNum() {
        return this.threadPool.getActiveCount();
    }

}

业务处理

import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

public class MoreThreadDealTest {
    public static void main(String[] args) {
        //模拟厂商并发处理
        for (int i = 1; i <= 4; i++) {
            int fiI = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    deal(fiI + "处理===");
                }
            }).start();
        }

    }

    public static void deal(String logHead) {
        final Map<Integer, String> resMap = new ConcurrentHashMap<>();//线程安全Map
        List<String> resList = new ArrayList<>();
        List<String> datas = new ArrayList<>();
        //模拟数据
        for (int q = 1; q <= 100; q++) {
            datas.add("" + q);
        }
        final CountDownLatch latch1 = new CountDownLatch(datas.size());
        long startTime = System.currentTimeMillis();
        try {
            for (int i = 0; i < datas.size(); i++) {
                int finalI = i;
                Runnable runnable = new Runnable() {
                    @SneakyThrows
                    @Override
                    public void run() {
                        try {
                            resMap.put(finalI, datas.get(finalI) + "处理了");
                            try {
                                Thread.sleep(3000);//业务耗时模拟
                            } catch (Exception e) {
                                System.out.println(logHead + "sleep异常" + e.getMessage());
                            }
                            System.out.println(logHead + "当前池线程个数==" + BusinessThreadPool.getInstance().getActivityNum());
                        } catch (Throwable e) {
                            System.out.println(logHead + "deal异常" + e.getMessage());
                        } finally {
                            latch1.countDown();
                        }
                    }
                };
                BusinessThreadPool.getInstance().execute(runnable);
            }
            System.out.println();
            latch1.await();
        } catch (Exception e) {
            System.out.print(logHead + "外层try异常了");
        } finally {
        }
        long endTime = System.currentTimeMillis();
        System.out.println();
        System.out.println(logHead + "---多线程异步处理耗时：" + (endTime - startTime));

        int size = resMap.size();
        for (int j = 0; j < size; j++) {
            resList.add(resMap.get(j));
        }
        System.out.println();
        System.out.println(logHead + "【处理后数据结构】" + JSON.toJSONString(resList));
        System.out.println();
        System.out.println(logHead + "===resList结果大小" + resList.size());
        long endTime2 = System.currentTimeMillis();
        System.out.println();
        System.out.println(logHead + "===总处理耗时：" + (endTime2 - startTime));
    }
}

日志实例

日志(部分)

2处理===当前池线程个数==200
2处理===当前池线程个数==200
2处理===当前池线程个数==200
2处理===当前池线程个数==200
2处理===当前池线程个数==200
2处理===当前池线程个数==200

3处理===当前池线程个数==200

4处理===当前池线程个数==139
4处理===当前池线程个数==138
4处理===当前池线程个数==137
4处理===当前池线程个数==136
4处理===当前池线程个数==135

4处理===---多线程异步处理耗时：6053

1处理===当前池线程个数==135
1处理===当前池线程个数==134
1处理===当前池线程个数==133
1处理===当前池线程个数==32

1处理===---多线程异步处理耗时：6069

2处理===当前池线程个数==31
2处理===当前池线程个数==30
2处理===当前池线程个数==29
2处理===当前池线程个数==28
2处理===当前池线程个数==27
2处理===当前池线程个数==26
2处理===当前池线程个数==14
2处理===当前池线程个数==13
2处理===当前池线程个数==12
2处理===当前池线程个数==11
2处理===当前池线程个数==10
2处理===当前池线程个数==9
3处理===当前池线程个数==8

3处理===---多线程异步处理耗时：6080

2处理===当前池线程个数==7
2处理===当前池线程个数==6
2处理===当前池线程个数==5
2处理===当前池线程个数==4
2处理===当前池线程个数==3
2处理===当前池线程个数==2
2处理===当前池线程个数==1

2处理===---多线程异步处理耗时：6080

2处理===【处理后数据结构】["1处理了","2处理了","3处理了","4处理了","5处理了","6处理了","7处理了","8处理了","9处理了","10处理了","11处理了","12处理了","13处理了","14处理了","15处理了","16处理了","17处理了","18处理了","19处理了","20处理了","21处理了","22处理了","23处理了","24处理了","25处理了","26处理了","27处理了","28处理了","29处理了","30处理了","31处理了","32处理了","33处理了","34处理了","35处理了","36处理了","37处理了","38处理了","39处理了","40处理了","41处理了","42处理了","43处理了","44处理了","45处理了","46处理了","47处理了","48处理了","49处理了","50处理了","51处理了","52处理了","53处理了","处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","处理了","90处理了","91处理了","92处理了","93处理了","94处理了","95处理了","96处理了","97处理了","98处理了","99处理了","100处理了"]

2处理======resList结果大小100

2处理======总处理耗时：6144
3处理===【处理后数据结构】["1处理了","2处理了","3处理了","4处理了","5处理了","6处理了","7处理了","8处理了","9处理了","10处理了","11处理了","12处理了","13处理了","14处理了","15处理了","16处理了","17处理了","18处理了","19处理了","20处理了","21处理了","22处理了","23处理了","24处理了","25处理了","26处理了","27处理了","28处理了","29处理了","30处理了","31处理了","32处理了","33处理了","34处理了","35处理了","36处理了","37处理了","38处理了","39处理了","40处理了","41处理了","42处理了","43处理了","44处理了","45处理了","46处理了","47处理了","48处理了","49处理了","50处理了","51处理了","52处理了","53处理了","处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","处理了","90处理了","91处理了","92处理了","93处理了","94处理了","95处理了","96处理了","97处理了","98处理了","99处理了","100处理了"]

3处理======resList结果大小100

3处理======总处理耗时：6145
1处理===【处理后数据结构】["1处理了","2处理了","3处理了","4处理了","5处理了","6处理了","7处理了","8处理了","9处理了","10处理了","11处理了","12处理了","13处理了","14处理了","15处理了","16处理了","17处理了","18处理了","19处理了","20处理了","21处理了","22处理了","23处理了","24处理了","25处理了","26处理了","27处理了","28处理了","29处理了","30处理了","31处理了","32处理了","33处理了","34处理了","35处理了","36处理了","37处理了","38处理了","39处理了","40处理了","41处理了","42处理了","43处理了","44处理了","45处理了","46处理了","47处理了","48处理了","49处理了","50处理了","51处理了","52处理了","53处理了","处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","处理了","90处理了","91处理了","92处理了","93处理了","94处理了","95处理了","96处理了","97处理了","98处理了","99处理了","100处理了"]

1处理======resList结果大小100

1处理======总处理耗时：6142
4处理===【处理后数据结构】["1处理了","2处理了","3处理了","4处理了","5处理了","6处理了","7处理了","8处理了","9处理了","10处理了","11处理了","12处理了","13处理了","14处理了","15处理了","16处理了","17处理了","18处理了","19处理了","20处理了","21处理了","22处理了","23处理了","24处理了","25处理了","26处理了","27处理了","28处理了","29处理了","30处理了","31处理了","32处理了","33处理了","34处理了","35处理了","36处理了","37处理了","38处理了","39处理了","40处理了","41处理了","42处理了","43处理了","44处理了","45处理了","46处理了","47处理了","48处理了","49处理了","50处理了","51处理了","52处理了","53处理了","处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","处理了","90处理了","91处理了","92处理了","93处理了","94处理了","95处理了","96处理了","97处理了","98处理了","99处理了","100处理了"]

4处理======resList结果大小100

4处理======总处理耗时：6150

解读

因篇幅问题不能全部显示，请点此查看更多更全内容

查看全文