没想到,这么简单的线程池用法,深藏这么多坑

又又又踩坑了

生产有个对账系统，每天需要从渠道端下载对账文件，然后开始日终对账。这个系统已经运行了很久，前两天突然收到短信预警，没有获取渠道端对账文件。

本以为又是渠道端搞事情，上去一排查才发现，所有下载任务都被阻塞了。再进一步排查源码，才发现自己一直用错了线程池某个方法。

慎用 Executors 组件

private ExecutorService threadPool = Executors.newFixedThreadPool(10);

/**
 * 批量下载对账文件
 *
 * @return
 */
@RequestMapping("/batchDownload")
public String batchDownload() {
    
    // 模拟下载 100w 个文件
    for (int i = 0; i < 1000000; i++) {
        threadPool.execute(() -> {
            // 随机休眠，模拟下载耗时
            Random random = new Random();
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    return "process";
}

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue());
}

/**
 * Creates a {@code LinkedBlockingQueue} with a capacity of
 * {@link Integer#MAX_VALUE}.
 */
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

复用线程池

public static ThreadPoolExecutor getThreadPool() {
    // 为了快速复现问题，故将线程池 核心线程数与最大线程数设置为 100
    return new ThreadPoolExecutor(100, 100, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(200));
}

@RequestMapping("/batchDownload")
public String batchDownload() {
    ExecutorService threadPool = ThreadPoolUtils.getThreadPool();

    // 模拟下载 100w 个文件
    for (int i = 0; i < 100; i++) {
        threadPool.execute(() -> {
            // 随机休眠，模拟下载耗时
            Random random = new Random();
            try {
                TimeUnit.SECONDS.sleep(random.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    return "process";
}

Spring 异步任务

@Async
public void async() throws InterruptedException {
    log.info("async process");
    Random random = new Random();
    TimeUnit.SECONDS.sleep(random.nextInt(100));
}

<task:executor id="myexecutor" pool-size="5"  />
<task:annotation-driven executor="myexecutor"/>

@Bean(name = "threadPoolTaskExecutor")
public Executor threadPoolTaskExecutor() {
    ThreadPoolTaskExecutor executor=new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setThreadNamePrefix("test-%d");
    // 其他设置
    return new ThreadPoolTaskExecutor();
}

@Async("threadPoolTaskExecutor")
public void xx() {
    // 业务逻辑
}

spring.task.execution.pool.core-size=10
spring.task.execution.pool.max-size=20
spring.task.execution.pool.queue-capacity=200

线程池方法使用不当

// 创建线程池
ExecutorService threadPool = ...
List> tasks = new ArrayList<>();
// 批量创建任务
for (int i = 0; i < 100; i++) {
    tasks.add(() -> {
        Random random = new Random();
        try {
            TimeUnit.SECONDS.sleep(random.nextInt(100));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "success";
    });
}
// 执行所有任务
List> futures = threadPool.invokeAll(tasks);
// 获取结果
for (Future future : futures) {
    try {
        future.get();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }
}

总结

有道无术，术可成；有术无道，止于术

欢迎大家关注Java之道公众号

好文章，我在看❤️