多线程应用

一、定义线程池

@Configuration
@EnableAsync
public class ThreadPoolConfig implements AsyncConfigurer {

    @Bean(Constant.THREAD_BEAN_NAME)
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Constant.THREAD_CORE_POOL_SIZE);
        executor.setMaxPoolSize(Constant.THREAD_MAX_POOL_SIZE);
        executor.setQueueCapacity(Constant.THREAD_QUEUE_CAPACITY);
        executor.setKeepAliveSeconds(Constant.THREAD_KEEP_ALIVE_TIME);
        executor.setThreadNamePrefix(Constant.THREAD_NAME_PREFIX);
        // 线程池对拒绝任务的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 初始化
        executor.initialize();
        return executor;
    }
}

二、引用线程池

 public class ExecuteTaskServiceImpl implements ExecuteTaskService {

@Autowired
 @Qualifier(value = Constant.THREAD_BEAN_NAME)
   
 private ThreadPoolTaskExecutor poolTaskExecutor;

  @Override
    public ResponseEntity<ResultInfo<RunTimeProcessCountStatisticVO>> runProcessCountStatistics(ProcessCountStatisticDTO processCountStatisticDTO) {
        AtomicInteger executionCompleteCount = new AtomicInteger();
        AtomicInteger executionInProcessCount = new AtomicInteger();

  try {
                    List<HistoricProcessInstance> list = Collections.synchronizedList(historicProcessInstanceList);
                    CountDownLatch latch = new CountDownLatch(list.size());
                    list.forEach(item -> poolTaskExecutor.execute(() -> multiHandleRunTimeProcess(item, historicTaskInstancesList, executionCompleteCount, executionInProcessCount, endDate, latch)));
                    latch.await();
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
}


    private void multiHandleRunTimeProcess(HistoricProcessInstance historicProcessInstance, Set<String> historicTaskInstancesList,AtomicInteger executionCompleteCount, AtomicInteger executionInProcessCount, Date endDate, CountDownLatch latch) {
            if (historicProcessInstance.getEndTime() != null && historicProcessInstance.getEndTime().before(endDate) && historicTaskInstancesList.contains(historicProcessInstance.getId())) {
                executionCompleteCount.getAndIncrement();
            } else if (historicProcessInstance.getEndTime() == null && historicTaskInstancesList.contains(historicProcessInstance.getId())) {
                executionInProcessCount.getAndIncrement();
            }
        latch.countDown();
    }

}

注意：当统计数量时要用保证数字的原子性，使用AtomicInteger,也可使用同步锁（synchronized）将逻辑锁住

/**
 * 任务数量统计多线程
 */
private void countDataMulHandle(TaskListDTO item, TaskCountStatisticVO vo, CountDownLatch latch) {
    int count = taskCountHandle(item).size();
    synchronized (this) {
        if (item.getTaskFilter().equals(Constant.STR_ALL)) {
            vo.setAll(count);
        } else if (item.getTaskFilter().equals(Constant.STR_SELF)) {
            vo.setSelf(count);
        } else {
            vo.setNotice(count);
        }
    }
    latch.countDown();
}