SpringBoot使用线程池实现异步批量处理任务

模拟批处理大量数据

@Slf4j
@Component
public class TestFutureService {

    @Autowired
    private TestFutureServiceImpl testFutureServiceImpl;

    /**
     *  多线程的优势：多核CPU使用多线程可以提高CPU的利用率（单核CPU不行，反而降低），可以实现异步调用。
     *
     *  （1）单核CPU同一时间只能处理一个线程，但是速度非常快，造成并行的假象（并发：交替轮流使用资源）
     *      因为一个CPU一次只能执行一条指令，使用多线程或多进程任务，这些任务其实都是并发执行的，操作系统会不断的切换多个任务。
     *      并发：指在同一个时刻做不止一件事情，比如数据库处理请求时，接受了第一个请求但未处理完成，此时也可以接受第二个请求。两个请求任务在处理时间节点上可以有交集。
     *  （2）多核CPU或CPU采用超线程技术的话，每个核心处理一个线程。（并行：多人分工同时处理任务）
     *      并行：指将耗时长的任务，拆解成多个子任务分配到多线程上，并发的执行。
     */

    @PostConstruct
    public void run(){
        runTask();
    }

    public void runTask() {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        log.info("开始执行批量处理任务...");
        List<String> dataList = new ArrayList<>();
        for (int i = 0; i < 50000; i++) {
            dataList.add(IdUtils.fastSimpleUUID());
        }
        // 每个线程处理多少条数据
        int capacity = 1000;
        // 计算得出所需的任务数量
        int count = NumberUtil.round(NumberUtil.div(dataList.size(), capacity), 0, RoundingMode.UP).intValue();
        CountDownLatch latch = new CountDownLatch(count);
        // 线程返回的结果集
        List<Future<List<String>>> futureList = new ArrayList<>();
        List<String> subList = new ArrayList<>(capacity);
        int k = 1;
        for (String id : dataList) {
            if (subList.size() >= capacity) {
                futureList.add(testFutureServiceImpl.doTask(subList, latch, k++));
                subList = new ArrayList<>(capacity);
            }
            subList.add(id);
        }
        if(subList.size() > 0){
            futureList.add(testFutureServiceImpl.doTask(subList, latch, k));
        }
        log.info("任务个数：" + count +"|"+ futureList.size());
        try {
            // 方式一：使用CountDownLatch计数器，等待所有任务执行完成
            latch.await();
            for (Future<List<String>> future : futureList) {
                List<String> pList = future.get();
                System.out.println("返回结果：" + pList.size());
            }
            // 方式二：使用轮循方式检查异步任务是否执行完成，特点：不需要计数器
            /**
            boolean bol = true;
            while (bol){
                for (Future<List<String>> future : futureList) {
                    if(future.isDone()){
                        bol = false;
                    } else {
                        bol = true;
                        break;
                    }
                }
                ThreadUtil.sleep(500);
            }
            */
            log.info("任务完成!");
        } catch (InterruptedException e) {
            log.error("处理线程中断异常", e);
        } catch (ExecutionException e) {
            log.error("处理执行异常", e);
        } catch (Exception e) {
            log.error("其它异常", e);
        }
        stopWatch.stop();
        log.info("任务执行耗时(秒)：" + stopWatch.getTotalTimeSeconds());
    }
}

定义使用线程池异步处理实现类

@Slf4j
@Component
public class TestFutureServiceImpl {

    // 需要自定义threadPoolTaskExecutor线程池配置，自行实现
    @Async("threadPoolTaskExecutor")
    public Future<List<String>> doTask(List<String> list, CountDownLatch latch, int k) {
        log.info("开始执行任务：" + k + "|" + list.size());
        ThreadUtil.sleep(1, TimeUnit.SECONDS);
        latch.countDown();
        return new AsyncResult<>(list);
    }
}