实践 Java 如何创建安全的线程池

 关注
发布于: 2020 年 06 月 14 日
《阿里巴巴Java开发手册》关于线程池的创建，有如下规则：
【强制】线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
说明：Executors 返回的线程池对象的弊端如下：
1）FixedThreadPool 和 SingleThreadPool:
允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
2）CachedThreadPool 和 ScheduledThreadPool:
允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。
﻿
如何理解并实践这条规则，创建没有OOM风险的、安全的线程池呢？以下主要分三个部分展开：
1、测试验证使用Executors创建的线程池为什么不安全
2、实现创建安全的CachedThreadPool、FixedThreadPool 、SingleThreadPool线程池，并测试验证
3、实现创建安全的ScheduledThreadPool线程池，并测试验证
1、测试验证使用Executors创建的线程池为什么不安全除了从源码角度理解，我们也可以通过编写测试用例，验证为什么使用 Executors 去创建线程池不安全，从而加深对规则的感性认识。这里我们构造了SleepTask任务执行类，按需sleep指定的毫秒数，模拟线程任务执行。
public class SleepTask implements Runnable {
    private final long sleepMilliSeconds;
    private final int taskId;
    public SleepTask(long sleepMilliSeconds, int taskId) {
        this.sleepMilliSeconds = sleepMilliSeconds;
        this.taskId = taskId;
    }
    public void run() {
        System.out.println("--------Start task " + taskId);
        try {
            Thread.sleep(sleepMilliSeconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("--------Finish task " + taskId);
    }
}
通过ExecutorsTester测试类，模拟了大量线程任务进入线程池执行/排队的场景，观察CachedThreadPool、FixedThreadPool和ScheduledThreadPool触发OOM时的情况，因SingleThreadPool与FixedThreadPool实现类似，可以简单认为是固定线程数为1的特殊FixedThreadPool，故略去相关代码。
public class ExecutorsTester {
    @Test
    public void testNewCachedThreadPool() {
        ExecutorService executorService = Executors.newCachedThreadPool();
        executeSleepTaskLoop(executorService);
    }
    @Test
    public void testNewFixedThreadPool() {
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        executeSleepTaskLoop(executorService);
    }
    @Test
    public void testNewScheduledThreadPool() {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);
        try {
            for(int i = 0; i < 100000; i++) {
                executorService.schedule(new SleepTask(2000, i), 1, TimeUnit.SECONDS);
                System.out.println(executorService.toString());
            }
        } catch (OutOfMemoryError e) {
            System.out.println(e.toString());
            System.exit(1);
        }
    }
    private void executeSleepTaskLoop(ExecutorService executorService) {
        try {
            for(int i = 0; i < 100000; i++) {
                executorService.submit(new SleepTask(3000, i));
                System.out.println(executorService.toString());
            }
        } catch (OutOfMemoryError e) {
            System.out.println(e.toString());
            System.exit(1);
        }
    }
}
测试用例执行时，限制了最大堆内存（-Xmx2m），以便较快触发OOM异常。
CachedThreadPool、FixedThreadPool和ScheduledThreadPool测试用例执行结果如下：
﻿
﻿
细心的你可能会发现，跟规则描述稍有不同，ScheduledThreadPool最终OOM时，打印出来的线程数并不大，反而跟FixedThreadPool类似，有大量的排队任务。通过查看源码发现，因为ScheduledThreadPool使用了自定义的DelayedWorkQueue，虽然初始队列大小是16，但会动态扩容至最大Integer.MAXVALUE，实际上也是无界工作队列。所以虽然配置的线程数量是Integer.MAXVALUE，但其实是用不到的，OOM的直接原因是无界队列堆积大量请求，这里《阿里巴巴Java开发手册》的描述值得推敲。
    static class DelayedWorkQueue extends AbstractQueue<Runnable>
            implements BlockingQueue<Runnable> {
        //只展示部分关键源码
        private static final int INITIAL_CAPACITY = 16;
        private void grow() {
            int oldCapacity = queue.length;
            int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
            if (newCapacity < 0) // overflow
                newCapacity = Integer.MAX_VALUE;
            queue = Arrays.copyOf(queue, newCapacity);
        }
        public int remainingCapacity() {
            return Integer.MAX_VALUE;
        }
    }
2、实现创建安全的CachedThreadPool、FixedThreadPool 、SingleThreadPool线程池，并测试验证下面我们按规则要求尝试创建安全的线程池，其中CachedThreadPool、FixedThreadPool 、 SingleThreadPool通过ThreadPoolExecutor 的方式创建都比较容易，可通过Guava的ThreadFactoryBuilder统一给线程池中的线程命名。
public class SafeExecutors {
    public static ExecutorService newCachedThreadPool(int corePoolSize, int maxPoolSize, int queueCapacity, String threadNamePrefix) {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-%d").build();
        BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<Runnable>(queueCapacity);
        return new ThreadPoolExecutor(corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS, blockingQueue, namedThreadFactory);
    }
    public static ExecutorService newFixedThreadPool(int corePoolSize, int queueCapacity, String threadNamePrefix) {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(threadNamePrefix + "-%d").build();
        BlockingQueue<Runnable> blockingQueue = new LinkedBlockingQueue<Runnable>(queueCapacity);
        return new ThreadPoolExecutor(corePoolSize, corePoolSize, 60L, TimeUnit.SECONDS, blockingQueue, namedThreadFactory);
    }
    public static ExecutorService newSingledThreadPool(int queueCapacity, String threadNamePrefix) {
        return newFixedThreadPool(1, queueCapacity, threadNamePrefix);
    }
}
测试验证也很简单，用例执行结果符合预期。
public class SafeExcutorsTester {
    @Test(expected = RejectedExecutionException.class)
    public void testSafeNewCachedThreadPool() {
        ExecutorService executorService = SafeExecutors.newCachedThreadPool(10, 100, 50, "test");
        executeSafeSleepTaskLoop(executorService);
    }
  
    @Test(expected = RejectedExecutionException.class)
    public void testSafeNewFixedThreadPool() {
        ExecutorService executorService = SafeExecutors.newFixedThreadPool(20, 50, "test");
        executeSafeSleepTaskLoop(executorService);
    }
  
    @Test(expected = RejectedExecutionException.class)
    public void testSafeNewSingledThreadPool() {
        ExecutorService executorService = SafeExecutors.newSingledThreadPool(50, "test");
        executeSafeSleepTaskLoop(executorService);
    }
  
    private void executeSafeSleepTaskLoop(ExecutorService executorService) {
        try {
            for(int i = 0; i < 100000; i++) {
                executorService.submit(new SleepTask(3000, i));
                System.out.println(executorService.toString());
            }
        } catch (RejectedExecutionException e) {
            System.out.println(e.toString());
            throw e;
        }
    }
}
﻿
3、实现创建安全的ScheduledThreadPool线程池，并测试验证实现创建安全的ScheduledThreadPool就有点麻烦了，因为ScheduledThreadPoolExecutor
虽然继承了ThreadPoolExecutor，但提供的构造方法没办法定制最大线程数和工作队列。
public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    //只展示部分关键源码
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue());
    }
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), threadFactory);
    }
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), handler);
    }
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                new DelayedWorkQueue(), threadFactory, handler);
    }
}
虽然可以通过super.setMaximumPoolSize方法重置最大线程数，但工作队列仍然不好处理，能不能也替代为有界的LinkedBlockingQueue呢？通过反射机制技术上是能做到的，但是细想意义不大，因为ScheduledThreadPoolExecutor定时任务执行的功能主要就是通过DelayedWorkQueue完成的，牺牲了定时功能ScheduledThreadPool就没有使用的意义了。
既然不能替换DelayedWorkQueue，那怎么解决无界队列可能引发的OOM呢？首先想到的就是定制DelayedWorkQueue为有界队列，但DelayedWorkQueue不是public的，也没有提供相关定制接口，难以实现。
还是从源码里找答案，发现ScheduledThreadPoolExecutor提供的关于DelayedWorkQueue队列操作的唯一public方法getQueue，是直接使用的父类的getQueue方法返回BlockingQueue，我们能不能间接的，在每次添加任务到队列前，增加检查队列长度的操作，人工控制工作队列的最大长度呢？ 
    public BlockingQueue<Runnable> getQueue() {
        return super.getQueue();
    }
﻿
因此尝试如下解决方案，自定义SafeScheduledThreadPool继承ScheduledThreadPool，覆写需要用到的定时方法，在调用父类的方法前，增加判断队列长度是否达到了自定义的上限，如果是就抛出RejectedExecutionException异常，避免队列无限增长。
public class SafeScheduledThreadPool extends ScheduledThreadPoolExecutor {
    protected final int queueCapacity;
    public SafeScheduledThreadPool(int corePoolSize, int maxPoolSize, int queueCapacity, ThreadFactory threadFactory) {
        super(corePoolSize, threadFactory);
        super.setMaximumPoolSize(maxPoolSize);
        this.queueCapacity = queueCapacity;
    }
    @Override
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        checkQueueSize();
        return super.schedule(command, delay, unit);
    }
    //省略覆写ScheduledThreadPoolExecutor中其他定时任务方法，实现同上
    private void checkQueueSize() {
        if (getQueue().size() >= queueCapacity) {
            throw new RejectedExecutionException("DelayedWorkQueue size exceed capacity: " + queueCapacity);
        }
    }
}
﻿
经过测试完美达到预期效果。
public class SafeExcutorsTester {
    @Test(expected = RejectedExecutionException.class)
    public void testSafeNewScheduledThreadPool() {
        ScheduledExecutorService scheduledExecutorService = SafeExecutors.newScheduledThreadPool(10, 100, 50, "test");
        try {
            for(int i = 0; i < 100000; i++) {
                scheduledExecutorService.schedule(new SleepTask(2000, i), 1, TimeUnit.SECONDS);
                System.out.println(scheduledExecutorService.toString());
            }
        } catch (RejectedExecutionException e) {
            System.out.println(e.toString());
            throw e;
        }
    }
}
﻿
﻿
总结通过动手测试更容易理解线程池的运行机制，以及编码规范的初衷，另外编码规范可能很精炼，自己实践时遇到问题还是要从源码里找答案，多尝试多验证，不浅尝辄止最终定有收货。 