具有小队列容量的 ThreadPoolTask​​Executor 阻塞调用线程？

Posted 2023-03-15

【中文标题】具有小队列容量的 ThreadPoolTask​​Executor 阻塞调用线程？

【英文标题】：ThreadPoolTaskExecutor with small queue capacity blocking calling thread?

【发布时间】：2021-08-08 22:31:02

【问题描述】：

抱歉，格式问题...

我已经尝试理解这一点大约 4 个小时了。基本上我有一个方法会调用一个使用 ThreadPoolTask​​Executor 的私有方法。

 public ThreadPoolTaskExecutor someTaskExecutor() 
    final ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(0);
    taskExecutor.setMaxPoolSize(3);
    taskExecutor.setKeepAliveSeconds(60);
    taskExecutor.setQueueCapacity(3);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;

这就是我使用它的方式。不要介意“DTO”和“Obj”，只需将其设为占位符

   @Override
public void processMessage(final DTO dto) throws Exception 
    log.info("MainThread BEFORE [] : [] - []", dto.getTaskId(), Thread.currentThread().getName(), Thread.currentThread().getId());
    final Object obj = save(dto);
    log.info("MainThread SAVED [] :  [] - []", dto.getTaskId(), Thread.currentThread().getName(), Thread.currentThread().getId());
    generateSomething(obj);
    log.info("MainThread AFTER [] :  [] - []", dto.getTaskId(), Thread.currentThread().getName(), Thread.currentThread().getId());


private void generateSomething(final Object obj) 
    someTaskExecutor.execute(() ->  
        log.info("thread START [] : [] - []", obj.getTaskId(), Thread.currentThread().getName(), Thread.currentThread().getId());
        //SOME API CALL THAT TAKES 3 second
        log.info("thread DONE  [] : [] - []", obj.getTaskId(), Thread.currentThread().getName(), Thread.currentThread().getId());

    );

以我的threadpooltaskexecutor的当前设置，当我对main方法有10个并发调用（产生10个主线程）时，只有6个线程成功通过调用私有方法后调用的“MainThread AFTER”有 threadpooltaskexecutor 执行。

目前我很迷茫，我的理解是主/调用线程不应该受到线程池任务执行器产生的新线程的影响，为什么它会直接受到队列容量的影响。如果我将队列容量设为 5 而不是 3，则 8 个主线程将正常通过。如果我不设置它，所有主线程都会正常通过。

感谢您的帮助，再次抱歉格式化。

【参考方案1】：

目前我很迷茫，我的理解是主/调用线程不应该受到线程池任务执行器产生的新线程的影响，为什么它会直接受到队列容量的影响。如果我将队列容量设为 5 而不是 3，则 8 个主线程将正常通过。如果我不设置它，所有主线程都会正常通过。

您正在设置一个ThreadPoolTaskExecutor（实际上是ThreadPoolExecutor），它最多有3 个线程，队列限制也为3。如果你向这个 executor 提交了 10 个作业并且这些作业需要一段时间才能完成，那么 3 个任务将开始运行，3 个将进入队列，当你尝试提交第 7 个任务时，调用者将阻塞。这就是它的工作原理。如果您增加排队任务的数量或增加可以处理这些任务的线程数量，那么调用者将稍后阻塞。

为什么阻止很重要？假设您有一个应用程序需要完成 1,000,000 个工作。您不能生成 1,000,000 个线程，因此您必须将作业排队。如果作业在内存中很大，您可能会用完容纳那么多的堆空间——或者考虑扩展到 1 亿个。通过阻塞，系统在保持高吞吐量的同时保持其线程和内存使用率低。您希望将一些作业排队，以便线程可以轻松地将它们从队列中取出并开始工作，但不要太多，以免占用应用程序所需的内存资源。