支持生产阻塞的Java线程池
通常来说，生产任务的速度要大于消费的速度。一个细节问题是，队列长度，以及如何匹配生产和消费的速度。
一个典型的生产者-消费者模型如下：
 |
对于一般的生产快于消费的情况。当队列已满时，我们并不希望有任何任务被忽略或得不到执行，此时生产者可以等待片刻再提交任务，更好的做法是，把生产者阻塞在提交任务的方法上，待队列未满时继续提交任务，这样就没有浪费的空转时间了。阻塞这一点也很容易，BlockingQueue就是为此打造的，ArrayBlockingQueue和LinkedBlockingQueue在构造时都可以提供容量做限制，其中LinkedBlockingQueue是在实际操作队列时在每次拿到锁以后判断容量。
更进一步，当队列为空时，消费者拿不到任务，可以等一会儿再拿，更好的做法是，用BlockingQueue的take方法，阻塞等待，当有任务时便可以立即获得执行，建议调用take的带超时参数的重载方法，超时后线程退出。这样当生产者事实上已经停止生产时，不至于让消费者无限等待。
于是一个高效的支持阻塞的生产消费模型就实现了。
等一下，既然J.U.C已经帮我们实现了线程池，为什么还要采用这一套东西？直接用ExecutorService不是更方便？
我们来看一下ThreadPoolExecutor的基本结构：
 |
但问题在于，即便你在构造ThreadPoolExecutor时手动指定了一个BlockingQueue作为队列实现，事实上当队列满时，execute方法并不会阻塞，原因在于ThreadPoolExecutor调用的是BlockingQueue非阻塞的offer方法：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
这时候就需要做一些事情来达成一个结果：当生产者提交任务，而队列已满时，能够让生产者阻塞住，等待任务被消费。
关键在于，在并发环境下，队列满不能由生产者去判断，不能调用ThreadPoolExecutor.getQueue().size()来判断队列是否满。
线程池的实现中，当队列满时会调用构造时传入的RejectedExecutionHandler去拒绝任务的处理。默认的实现是AbortPolicy，直接抛出一个RejectedExecutionException。
几种拒绝策略在这里就不赘述了，这里和我们的需求比较接近的是CallerRunsPolicy，这种策略会在队列满时，让提交任务的线程去执行任务，相当于让生产者临时去干了消费者干的活儿，这样生产者虽然没有被阻塞，但提交任务也会被暂停。
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a <tt>CallerRunsPolicy</tt>.
*/
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
但这种策略也有隐患，当生产者较少时，生产者消费任务的时间里，消费者可能已经把任务都消费完了，队列处于空状态，当生产者执行完任务后才能再继续生产任务，这个过程中可能导致消费者线程的饥饿。
参考类似的思路，最简单的做法，我们可以直接定义一个RejectedExecutionHandler，当队列满时改为调用BlockingQueue.put来实现生产者的阻塞：
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if (!executor.isShutdown()) {
try {
executor.getQueue().put(r);
} catch (InterruptedException e) {
// should not be interrupted
}
}
}
};
这样，我们就无需再关心Queue和Consumer的逻辑，只要把精力集中在生产者和消费者线程的实现逻辑上，只管往线程池提交任务就行了。
相比最初的设计，这种方式的代码量能减少不少，而且能避免并发环境的很多问题。当然，你也可以采用另外的手段，例如在提交时采用信号量做入口限制等，但是如果仅仅是要让生产者阻塞，那就显得复杂了。
相关文章
在本文里我们给大家整理了关于Java常用排序算法以及实例代码分析，需要的朋友们跟着学习下。2019-03-03
这篇文章主要介绍了模拟Java多线程环境下的死锁，文章介绍一些死锁的产生条件的相关资料，具有一定的参考价值，需要的小伙伴可以参考一下，希望对你有所帮助2021-12-12
这篇文章主要介绍了消息中间件详解以及比较选择,本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下2021-08-08
mybatis-flex与springBoot整合的实现示例
Mybatis-flex提供了简单易用的API,开发者只需要简单的配置即可使用，本文主要介绍了mybatis-flex与springBoot整合，具有一定的参考价值，感兴趣的可以了解一下2024-01-01
本文主要介绍了SpringBoot集成百度AI实现人脸识别的项目实践，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-05-05
这篇文章主要介绍了Java 对称加密使用DES / 3DES / AES 这三种算法分别实现的相关资料,这里提供了实例代码，需要的朋友可以参考下2017-01-01
本篇文章主要介绍了springboot开启声明式事务的方法，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2017-05-05
在本篇文章里小编给各位整理的是关于SpringBoot整合Mybatis的知识点汇总，有兴趣学习的参考下。2020-02-02
这篇文章主要介绍了详解Java进阶知识注解，从注解的定义、元注解、自定义注解、注解实例这几个方面，让同学们更加深入的了解注解2021-04-04
这篇文章主要介绍了java的split方法使用示例,需要的朋友可以参考下2014-04-04
最新评论