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

方案设想

• 多线程异步“并行”处理待处理数据【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处理了","54处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","64处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","89处理了","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处理了","54处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","64处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","89处理了","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处理了","54处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","64处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","89处理了","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处理了","54处理了","55处理了","56处理了","57处理了","58处理了","59处理了","60处理了","61处理了","62处理了","63处理了","64处理了","65处理了","66处理了","67处理了","68处理了","69处理了","70处理了","71处理了","72处理了","73处理了","74处理了","75处理了","76处理了","77处理了","78处理了","79处理了","80处理了","81处理了","82处理了","83处理了","84处理了","85处理了","86处理了","87处理了","88处理了","89处理了","90处理了","91处理了","92处理了","93处理了","94处理了","95处理了","96处理了","97处理了","98处理了","99处理了","100处理了"]
  
  4处理======resList结果大小100
  
  4处理======总处理耗时：6150
  

• 解读

  1. resList结果大小100 --ConcurrentHashMap 数据结构ok
  2. 总处理耗时 --模拟4个厂商，每个厂商100个，每个任务3秒，耗时6秒ok
  3. 前池线程个数 --动态重复使用和衰减，可行