Java 线程池有限大小工作队列 - 不丢弃任务的实现

我们在创建 Java 线程池,无论是用 Executors, ThreadPoolExecutor, 还是 Spring 的 ThreadPoolTaskExecutor, 如果不指定工作队列的大小的话,默认为 Integer.MAX_VALUE(2147483647), 基本不会把它爆满,但是在许多的任务要执行时大量 Runnable 对象的创建却足以把内存撑爆掉。所以才有必要使用一个有限大小的工作队列,如 5000, 再配上 RejectedExecutionHandler(DiscardOldestPolicy, DiscardPolicy, 或 CallerRunsPolicy)。前两种策略会主动放弃最旧最新的任务,一般不是我们想要的,CallerRunsPolicy 还能主动发挥任务提交者的计算能力,是一种不错的选择(只可能会发生工作队列太小且提交者执行的任务太忙时产生线程池一时的空闲。

所以总结起来我们可以有以下几种实现

直接使用 CallerRunsPolicy

在工作队列满时有效利用提交任务的线程,不让它闲着,这种实现最简单, 像下面那样声明线程池

自定义工作队列

在自定义工作队列中让提交任务的线程一直等待 -- 主动的转换立即返回的 offer() 调用转换为持续等待的 put() 操作

这就是在之前 理解 Java 线程池 ThreadPoolExecutor 一文中介绍的实现代码,代码是

实现 RejectedExecutionHandler 接口

接受任务提交失败,然后转换为 put() 等待. 这种方式本质上去前一种是一样的,只是被动的把立即返回的 offer() 调用转换为持续等待的 put() 操作

它要比前一种方法稍微简单些,我们可以测试一下效果

每个任务会休眠半秒钟,肯定会造成工作队列满而提交失败的情况,看输出最后结果是

......
everRejectedCount: 40, executedCount: 100

中间有许多 rejected 的输出,总之所有任务都得到正常执行了。

按需选择吧。

本文链接 https://yanbin.blog/java-threadpool-bounded-workqueue-block-wait/, 来自 隔叶黄莺 Yanbin Blog

[版权声明] Creative Commons License 本文采用 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 进行许可。

Subscribe
Notify of
guest

2 Comments
Inline Feedbacks
View all comments
bbbush
6 months ago

can the submitter thread refresh outstanding SQS messages, rather than to fetch new messages?