Java并发之线程池

在实际开发中，并不会直接创建使用线程，而是使用池化思想，创建线程池。 降低资源消耗，提高响应速度，提高线程的管理性。

构造方法

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

• corePoolSize 核心线程数目（最多保留的线程数）
• maximumPoolSize 最大线程数目
• keepAliveTime 生存时间 - 针对救急线程
• unit 时间单位 - 整堆救急线程
• workQueue 阻塞队列
• threadFactory 线程工厂- 可以为线程创建时起名字
• handler 拒绝策略

核心线程数目+救急线程数目=最大线程数目。

任务优先使用核心线程，当核心线程用完后进入阻塞队列排队，当阻塞队列满时用救急线程执行此时进来的任务，如果救急线程用完了，则会执行拒绝策略。

• 线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务
• 当线程数达到corePoolSize，这时再加入任务，新加入的任务会被加入workQueue队列排队，直到有空闲的线程
• 如果队列选择了有界队列，那么任务超过了队列大小时，会创建maximumPoolSize - corePoolSize数目的线程来救急
• 如果线程到达maximumPoolSize仍然有新任务这时会执行拒绝策略。JDK提供了4种实现
  • AbortPolicy 让调用者抛出RejectedExecutionException异常，这是默认策略
  • CallerRunsPolicy让调用者运行任务
  • DiscardPolicy 放弃本次任务
  • DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
  • Dubbo实现，在抛出RejectedExecutionException异常之前会记录日志，并dump线程栈信息，方便定位问题
  • Netty实现，创建一个新线程来执行任务
• 当高峰过去后，超过corePoolSize的救急线程如果一段时间没有任务做，会结束，这个时间由keepAliveTime和unit控制

创建固定大小的线程池newFixedThreadPool

• 核心线程数=最大线程数（没有救急线程）
• 阻塞队列是无界的，可以放任意数量的任务
• 适用于任务量已知，相对耗时的任务
• 因为阻塞队列是无界的，当任务非常多的时候，容易OOM

        ExecutorService executorService = Executors.newFixedThreadPool(3, new ThreadFactory() {
            private AtomicInteger t = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "customThreadFactory-" + t.getAndIncrement());
            }
        });
        executorService.execute(() -> {
            System.out.println(Thread.currentThread() + "1");
        });
        executorService.execute(() -> {
            System.out.println(Thread.currentThread() + "2");
        });
        executorService.execute(() -> {
            System.out.println(Thread.currentThread() + "3");
        });

创建带缓存的线程池newCachedThreadPool

• 核心线程数是0，最大线程数是Integer.MAX_VALUE，救急线程的空闲生存时间是60s，意味着全部都是救急线程
• 队列采用了SynchronousQueue，它没有容量，没有线程来取是放不进去的
• 适合任务数比较密集，但每个任务执行时间比较短的情况
• 最大线程数是Integer.MAX_VALUE，可能会创建大量线程而OOM

创建单线程线程池newSingleThreadExecutor

希望多个任务排队执行。线程数固定为1，任务数多于1时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。

• 与自己创建的单线程执行任务区别？自己创建的任务失败了，后续没有补救，而线程池会新建线程，保证后续正常工作。
• 使用无界队列作为阻塞队列，当任务量很大的时候容易OOM

        ExecutorService executorService = Executors.newSingleThreadExecutor();
		//线程1发生了异常
        executorService.execute(() -> {
            System.out.println(Thread.currentThread() + "工作");
            int i = 1 / 0;
        });
		//创建了新的线程2继续工作
        executorService.execute(() -> {
            System.out.println(Thread.currentThread() + "工作");
        });

• 与newFixedThreadPool(1)区别？newSingleThreadExecutor()线程个数始终为1，不能修改

        ExecutorService threadPool = Executors.newFixedThreadPool(2);
        Future<String> submit = threadPool.submit(() -> "完成");
        System.out.println(submit.get());

一些API

        ExecutorService threadPool = Executors.newFixedThreadPool(2);

		//submit提交任务，用返回值Future获得任务执行结果
        Future<String> submit = threadPool.submit(() -> "完成");
        System.out.println(submit.get());

        //invokeAll，提交所有任务
        List<Future<String>> futures = threadPool.invokeAll(Arrays.asList(
                () -> {
                    System.out.println("开始1");
                    return "完成1";
                },
                () -> {
                    System.out.println("开始2");
                    return "完成2";
                }
        ));

        futures.forEach(t->{
            try {
                System.out.println(t.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });

        //invokeAny，任意一个任务完成就返回
        String str = threadPool.invokeAny(Arrays.asList(
                () -> {
                    System.out.println("开始1");
                    return "完成1";
                },
                () -> {
                    System.out.println("完成2");
                    return "完成2";
                }
        ));
        System.out.println(str);
		//关闭线程池。不会再接受新任务，已提交的任务会执行完，不会阻塞调用线程的执行
		threadPool.shutdown();
		//立即关闭线程池。已提交的任务返回，打断正在执行的线程。
    	threadPool.shutdownNow();

创建多少线程池合适

• 对于CPU密集型运算。通常采用CPU核数+1
• 对于I/O密集型运算（Web应用）。线程数=核数 * 期望CPU利用率 * 总时间(CPU计算时间+等待时间) / CPU计算时间

任务调度线程池newScheduledThreadPool

用于执行定时任务，但实际开发不会选择使用，一般使用quartz。