2020-02-01 2.7 线程池工具

本文是Java线程安全和并发编程知识总结的一部分。

2.7 线程池工具

JDK在java.util.concurrent包底下，通过Executor框架提供了多线程执行任务的完整工具包。

2.7.1 Executor框架

Executor接口衍生出 ExecutorService接口，提供了如下几类方法：

提交任务

submit：各种类型的提交任务的重载方法

执行任务：

execute：执行某个任务的方法

invokeAll：各种调用所有指定任务的重载方法

invokeAny：各种调用指定任务中一个任务的重载方法

执行任务：

shutdown()：不再接收新任务，等正在执行的任务结束后关闭Executor。

shutdownNow()：立即关闭Executor，放弃正在执行的任务。

isShutdown：是否已关闭

isTerminated：是否所有任务已完成

awaitTermination：申请关闭Executor后阻塞，直到所有任务完成，或超时，或当前线程中断。

Executor是和线程池一起工作的。Executor内部有任务队列和线程池；前者放置已提交待执行的任务；后者提供线程来执行任务。提交任务时，任务被接受并放在任务队列中；当有空闲线程时，任务队列中的一个任务移交给线程池中的一个线程执行。

通过分别控制任务队列和线程池的大小、任务移交方式，可以得到不同具体行为的Executor实现。Executor的子类非常多，不同的子类和线程池协同工作的方式不一样。

Executors工具类提供了最常用的几类Executor子类的工厂方法：

newFixedThreadPool(): 创建ThreadPoolExecutor；使用固定大小的线程池；使用LinkedBlockingQueue做工作队列，可接收无限任务。

newSingleThreadExecutor(): 创建FinalizableDelegatedExecutorService；使用只有一个线程的线程池，意味着永远只有一个线程在工作；使用LinkedBlockingQueue做工作队列，可接收无限任务。

newCachedThreadPool(): 创建ThreadPoolExecutor；使用无限大小的线程池；使用SynchronousQueue做工作队列，可接收无限任务并立即移交执行。

newSingleThreadScheduledExecutor(): 创建ScheduledExecutorService；使用只有一个线程的线程池，意味着只有一个线程在工作；使用DelayedWorkQueue做工作队列，以延迟或定时的方式来执行任务。

newScheduledThreadPool(): 创建ScheduledExecutorService；使用指定大小的线程池；使用DelayedWorkQueue做工作队列，以延迟或定时的方式来执行任务。

Executors工具类还提供了一系列创建Callable的工厂方法：

callable(): 一系列接收 Runnable，创建Callable的重载方法。

callable(): 一系列接收PrivilegedAction，创建Callable的重载方法。

privilegedCallable(): 一系列接收callable，创建PrivilegedCallable的重载方法。

Runnable和Callable的差别：

Runnable只执行一段逻辑，不返回数据；如果执行的逻辑需要返回数据，只能通过共享数据。
Callable返回一个Future，这是一个线程安全的阻塞的结果。调用Future.get()时，如果已经有计算结果，则返回结果集；如果尚未尚未计算完毕，则当前线程阻塞。

还有其他一些ExecutorService和Callable的子类，因为不常用，一般都是需要使用时，再去查阅API选用。

有一个特殊的工具类 ExecutorCompletionService，需要单独提到；它提供submit方法来提交任务；使用take方法按照任务完成先后顺序来获取计算结果，如果尚未有任务计算完成，则阻塞当前线程。
它内部使用Executor来执行任务，同时使用一个BlockingQueue按照任务完成的先后顺序来存放任务执行的结果Future。BlockingQueue的特性，保证了take方法的效果。
这是一个使用CompletionService的例子：

/**
 * @author xx
 * 2020年1月31日 下午4:08:45
 */
public class Sample11 {
    
    /**
     * 时间格式
     */
    private final DateTimeFormatter formater = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS", Locale.CHINESE);
    
    /**
     * 随机数生成器
     */
    private final Random random = new Random(5);
    
    /**
     * 使用固定线程池大小的Executor
     */
    private final ExecutorService exe = Executors.newFixedThreadPool(50);
    
    /**
     * 
     * 2020年2月3日 上午11:18:42 xx添加此方法
     * @param args
     */
    public static void main(String[] args) {
        Sample11 sample11 = new Sample11();
        try {
            sample11.startCalc();
        } finally {
            sample11.shutdown();
        }
    }
    
    /**
     * 使用CompletionService执行任务的例子
     * 2020年2月3日 上午11:18:42 xx添加此方法
     */
    public void startCalc() {
        // 使用无限任务队列且任务立即移交，10个执行线程的Executor
        CompletionService<String> compService = new ExecutorCompletionService<>(this.exe);
        for ( int i = 0; i <25; i++) {
            final String taskName = "task" + i;
            compService.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return Sample11.this.calc(taskName);
                }
            });
        }
        
        // 总共只有25个任务，因此等到25个结果后，就可以结束了，无需永远等待下一个结果被计算出来。
        for (int i = 0; i <25; i++) {
            Future<String> result;
            try {
                result = compService.take();
                System.out.println(result.get());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }
    }
    
    /**
     * 关闭Executor，否则主线程不会结束。
     * 2020年2月3日 下午12:13:15 xx添加此方法
     */
    public void shutdown() {
        this.exe.shutdown();
    }
    
    /**
     * 模拟业务计算，确保每个任务耗时不同。
     * 2020年2月3日 上午11:18:42 xx添加此方法
     */
    public String calc(String taskName) {
        System.out.println(taskName + "开始执行了");
        String startTime = this.instantFormat(Instant.now());
        
        // 模拟做业务操作。
        int randonMs = this.random.nextInt(1000);
        System.out.println(taskName + " 使用随机数 " + randonMs);
        
        try {
            Thread.sleep(randonMs);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        String endTime = this.instantFormat(Instant.now());
        
        return taskName +"从 " + startTime +" 开始执行，到 " + endTime + " 结束 ";
    }
    
    /**
     * 格式化时间为字符串 yyyy-MM-dd HH:mm:ss.SSS
     * 2020年2月3日 上午11:26:56 xx添加此方法
     * @param instant 待格式化的时间对象
     * @return
     */
    private String instantFormat(Instant instant) {
        LocalDateTime localDateTime = instant.atZone(ZoneId.ofOffset("", ZoneOffset.ofHours(8))).toLocalDateTime();
        return localDateTime.format(formater);
    }
}

上述接口和工具方法，构成了JDK提供的线程池工具。

2.7.2 线程池使用要点

线程池的使用，也有很多需要注意的地方。

2.7.2.1 线程饥饿死锁

当一个正在运行的线程，提交一个新任务给自己所在的线程池，并依赖该任务的完成才能继续进行时，如果线程池的大小不合适，导致新任务始终处于线程池的待执行任务列表中，那么就构成了一个饥饿死锁。

因为线程池中线程不足，所以新任务无法被执行；因为正在运行的线程需要等待新任务的执行结果，所以正在运行的线程无法执行完成后释放，导致线程更加不足。

2.7.2.2 为运行时间较长的任务设计线程池大小和超时时间

当有运行时间较长的任务时，必须使线程池足以容纳在一定时间内可能同时执行的运行时间较长的任务。

设置线程池大小时，需要对一定时间内可能出现的任务数有一定的预计，从而规划线程池的大小，使其足以容纳这些同时执行的执行时间较长的任务。否则会因为这些任务占据了大部分甚至全部线程，而导致运行时间较短的任务因为等待线程而变成性能低下。
对运行时间较长的任务，需要估算的合理的运行时长，并设定一定的超时策略，避免无限等待导致实时死锁。