如何实现线程池的生产消费平衡

Posted 2020-12-12 abel-huang

Java线程池Executor框架可以视为一个生产者消费者模型，但是如何正确保持生产者和消费者平衡，也就是向线程池投递任务的速度和线程池处理任务能达到一个平衡关系，被投递的任务能够及时的被处理，不会因为超过线程池负载出现任务丢失甚至导致程序崩溃。由于Executor框架默认提供了空闲线程回收机制，所以不用担心消费者闲置的情况，我们主要考虑的问题是生产者生产速度太大导致消费者消费能力跟不上的情况，当消费者达到负载极限时可以限制生产者的投递速度从而达到平衡效果。

线程池主要由三个部分组成：核心线程池，最大线程池，阻塞任务队列。我们知道线程池中线程流入的顺序是核心线程池，到阻塞任务队列，最后到最大线程池，如果线程池达到最大线程数后再进入的任务就会触发拒绝策略RejectPolicy，其中解决问题的关键方法就是阻塞任务队列和拒绝策略。

基于CallerRunsPolicy拒绝策略

ThreadPoolExecutor提供了四种拒绝策略的实现，其默认实现采用了AbortPolicy，当达到最大线程数后会触发一个RejectedExecutionExcep异常，本次任务也会丢弃，这显然不符合我们的需求。CallerRunsPolicy意思是如果线程池达到最大值，则会由当前线程来执行，也就是会由生产者线程(一般是主线程)来执行任务，由生产者替消费者分担压力，可以算是一个负反馈系统了。具体代码如下：

ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("caller-run-executor").build();
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,??10,??0,?TimeUnit.SECONDS,?new LinkedBlockingQueue<>(10),?threadFactory,?new ThreadPoolExecutor.CallerRunsPolicy());

那CallerRunsPolicy()究竟是怎么实现的，参考一下其源码：

  public static class CallerRunsPolicy implements RejectedExecutionHandler {
      
        public CallerRunsPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

核心代码就是实现RejectedExecutionHandler的rejectedExecution(Runnable r, ThreadPoolExecutor e)方法，这里我们可以看到直接调用了run()方法，也就是直接在生产者线程里面同步执行该任务。
使用这种方式可以达到我们平衡生产消费者的目的，但是也存在一个问题，那就是被线程池拒绝的任务可能会先于部分先提交的任务而执行，如果对任务执行顺序比较敏感的服务对这种方法需要有所考虑。

自定义RejectedExecutionHandler

通过自定义RejectedExecutionHandler也可以实现类似的需求：

ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("customer-pool-executor").build();
ThreadPoolExecutor executor = new ThreadPoolExecutor(5,??10,??0,??TimeUnit.SECONDS,?? new LinkedBlockingQueue<>(10),??threadFactory,??new CustomRejectedExecutionHandler();

CustomRejectedExecutionHandler实现如下

public  class CustomRejectedExecutionHandler implements RejectedExecutionHandler {??? 
    @Override??? 
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {??????? 
        try {??????????? 
            // offer改成put阻塞方法???????????
            executor.getQueue().put(r);??????? 
        } catch (InterruptedException e) {??????????? 
            e.printStackTrace();??????? 
        }??
}

在ThreadPoolExecutor的execute()方法实现中，BlockingQueue入队都是直接使用非阻塞的offer()方法，可以通过阻塞方法put()将任务重新入队，如果任务队列已经满了，那么生产者线程会直接阻塞，直到任务队列出现空闲位置，不过采用这种方案的前提是线程池定义时采用的是有界队列(当然这也是推荐的设置)。

重写execute(Runnable command)方法

上面说到execute()默认实现入队使用的是非阻塞方法offer()，如果直接重写execute()方法并改成阻塞方法那么可以达到同样的效果。不过需要注意一下，如果采用这种方式需要采用默认的AbortPolicy策略，通过捕获RejectedExecutionException检测是否已经达到线程池上线，这样做是因为我们为了“偷懒”，对于execute()方法的实现直接调用了其父类的实现，如果完全是自定义实现的则可以忽略这条事项。

ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("customer-policy-executor").build();
ThreadPoolExecutor executor = new CustomerThreadPoolExecutor(5,??10,??0,?TimeUnit.SECONDS,?new LinkedBlockingQueue<>(10),?threadFactory, new ThreadPoolExecutor.AbortPolicy());

CustomerThreadPoolExecutor实现如下

public class CustomerThreadPoolExecutor extends ThreadPoolExecutor {

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public CustomerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    @Override
    public void execute(Runnable command) {
        try {
            super.execute(command);
        } catch (RejectedExecutionException rx) {
            try {
                BlockingQueue<Runnable> queue = super.getQueue();
                queue.put(command);
            } catch (InterruptedException x) {
                super.getRejectedExecutionHandler().rejectedExecution(command, this);
            }
        }
    }
}

以上三种方法都可以达到线程池生产消费平衡的目的，具体采用哪种方案可以根据自己业务的实际情况采纳。