线程池之ThreadPoolExecutor使用

线程池之ThreadPoolExecutor使用

如图ThreadPoolExecutor的父类是AbstractExecutorService，AbstractExecutorService的父类是ExecutorService，ExecutorService的父类是Executor，所以ThreadPoolExecutor就相当于线程池的执行器

ThreadPoolExecutor共有四个构造方法，我们以参数最多的对其参数进行解释

corePoolSoze：核心线程数，最开始时的线程数量
maximumPoolSize：最大线程数量，当核心线程数不够时，允许拓展的线程数
keepAliveTime：线程最长的空闲时间，超过这个时间则把线程归还给 *** 作系统
unit：空闲时间的单位
workQueue：任务等待队列
defaultThreadFactory：线程创建工厂，默认产生的是defaultThreadFactory，可以自定义线程创建工厂，例如指定线程名称方便出错回溯
handler：拒绝策略，但线程数量不够，且等待队列满了的时候要执行的拒绝策略，默认有四种拒绝策略，也支持自定义。

1. Abort：抛异常
2. Discard：扔掉，不抛异常
3. DiscardOldest：扔掉排队时间最久的
4. CallerRuns：调用者处理服务

一般我们会自定义拒绝策略，把消息保存在消息中间里慢慢去消费。
下面来看JDK给我们提供了哪些默认的实现
SingleThreadPool

它的核心线程和最大线程数量都是1，为什么会有这样的线程池呢，因为线程池可以帮我们管理线程的生命周期，他有自己的任务队列，扔进去的任务是顺序执行。

public static void main(String[] args) {
	ExecutorService service = Executors.newSingleThreadExecutor();
	for(int i=0; i<5; i++) {
		final int finalI = i;
		service.execute(()->{
			System.out.println(finalI + " " + Thread.currentThread().getName());
		});
	}
}

FixedThreadPool

通过源码可以了解到它的核心线程数和最大线程数都是固定的，没有回收线程，所以keepAliveTime是0，用的等待队列是linkedBlockingQueue（在阿里规范中并不建议使用，因为容易出现oom）

public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Work work1 = new Work();
        Work work2 = new Work();
        Work work3 = new Work();

        executorService.submit(work1);
        executorService.submit(work2);
        executorService.submit(work3);
    }

    static class Work extends Thread {

        @Override
        public void run() {
            System.out.println("name:" + Thread.currentThread().getName());
        }
    }

CachedPool

通过源码可以得知，线程数量几乎不限制，来一个任务就创建个线程，线程超过60s是空闲状态时就会回收，因为没线程数量限制所以用的是SynchronousQueue，来一个任务就必须拿走不然提交任务时就会阻塞，适用于处理大量耗时短的任务，如Netty的NIO接收请求。

ScheduledPool

当我们调用newScheduledThreadPool的时候它还是返回的ScheduledThreadPoolExecutor，在ScheduledThreadPoolExecutor里调用了super，它的super还是ThreadPoolExecutor。

它用的任务队列是DelayedWorkQueue，DelayedWorkQueue是无界阻塞队列，它的队列元素只能在该元素的延迟已经结束才能被出队。
他的主要用法有三个：

public static void main(String[] args) {
        ScheduledExecutorService service = Executors.newScheduledThreadPool(3);
        System.out.println("start:" + System.currentTimeMillis());
        // 延时任务
        service.schedule(()->{
            System.out.println("schedule:" + System.currentTimeMillis() + " Name:" + Thread.currentThread().getName());
        }, 1, TimeUnit.SECONDS);

        // 循环任务，用上一次任务的开始时间计算下一次任务的开始时间
        service.scheduleAtFixedRate(() -> {
            System.out.println("Rate:" + System.currentTimeMillis() + " Name:" + Thread.currentThread().getName());
        }, 1, 1, TimeUnit.SECONDS);

        // 循环任务，用上一次任务的结束时间计算下一次任务的开始时间
        service.scheduleWithFixedDelay(() -> {
            System.out.println("Delay:" + System.currentTimeMillis() + " Name:" + Thread.currentThread().getName());
        }, 1, 1, TimeUnit.SECONDS);

    }

其中initialDelay是第一个任务延迟多少时间执行，period参数是间隔多少时间执行，unit是时间单位，通过这些参数可以灵活控制任务的执行时间。
自定义线程池

public class MyThreadPoolExecutor {

    public static void main(String[] args) throws IOException {
        ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4, 1000,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(2), new MyThreadFactory(), new MyRejectedExecutionHandler());

        for (int i = 0; i < 10; i++) {
            tpe.execute(new Task("name" + i));
        }
        try {
            // 让主线程阻塞
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    static class MyThreadFactory implements ThreadFactory {

        private final AtomicInteger i = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Thread-" + i);
        }
    }

    static class MyRejectedExecutionHandler implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            System.out.println(r.toString() + "rejected");
        }
    }

    static class Task extends Thread {

        private String name;

        Task(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                // 阻塞住更好看到效果
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(name + " start");
        }
    }
}

执行结果

总结：在实际生产环境中用自定义线程池会更好