JUC-线程池-蒲公英云

ThreadPoolExecutor

线程状态

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量。

状态名	高 3位	接收新任务	处理阻塞队列任务	说明
running	111	Y	Y
shutdown	000	N	Y	不会接收新任务，但会处理阻塞队列剩余任务
stop	001	N	N	会中断正在执行的任务，并抛弃阻塞队列任务
tidying	010			任务全执行完毕，活动线程为 0 即将进入终结
terminated	011			终结状态

这些信息存储在原子变量ctl中，将线程状态与线程数合二为一，这样就可以用一次cas操作进行赋值。

// c 为旧值， ctlOf 返回结果为新值
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
// rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
private static int ctlOf(int rs, int wc) {
     return rs | wc; }

构造方法

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

选取一个参数最多的构造方法，看看创建的时候都需要传入什么参数
在这里插入图片描述

corePoolSize 核心线程数
maximumPoolSize 最大线程数
keepAliveTime 救急线程存活时间
unit 时间单位
workQueue 阻塞队列
threadFactory 线程工厂
handler 拒绝策略

工作方式就是当线程池中还没有线程时，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。然后任务数达到核心线程数时就会把新提交的任务交给阻塞队列，如果阻塞队列是有界队列，当队列满了之后就会创建救急线程，救急线程数等于最大线程数减去核心线程数，救急线程满了之后就会执行解决策略。
拒绝策略jdk提供了四种实现方式：

AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
CallerRunsPolicy 让调用者运行任务
DiscardPolicy 放弃本次任务
DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之

再了解一下其他框架的拒绝策略实现：

Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
Netty 的实现，是创建一个新线程来执行任务
ActiveMQ 的实现，带超时等待（60s）尝试放入队列

当线程池中的救急线程没有任务执行后，就会存活keepAliveTime和unit控制的时间就会销毁。

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

核心线程数就是最大线程数
阻塞队列是无界的，可以放任意数量任务

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

核心线程数为0，救急线程为整数最大值，生存时间为60秒，意味着全是救急线程。
阻塞队列采用SynchronousQueue 实现，特点是没有容量，没有线程来取是放不进去的

SynchronousQueue integers = new SynchronousQueue<>();
```
    new  Thread(()->{
```

            try {
                System.out.println("放入任务一");
                integers.put(1);
                System.out.println("任务一放入完成");
                System.out.println("放入任务二");
                integers.put(2);
                System.out.println("任务二放入完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t1").start();
        sleep(1);
        new Thread(() -> {
            try {
                System.out.println("拿任务一");
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t2").start();
        sleep(1);
        new Thread(() -> {
            try {
                System.out.println("拿任务二");
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t3").start();

在这里插入图片描述
通过例子能体会到没有线程来取是放不进去的
使用场景适合任务数比较密集，但每个任务执行时间较短的情况

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

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

自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作
FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法，所以线程数始终是一不能修改。

任务提交

// 执行任务
void execute(Runnable command);
// 提交任务 task，用返回值 Future 获得任务执行结果
<T> Future<T> submit(Callable<T> task);
// 提交 tasks 中所有任务
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
// 提交 tasks 中所有任务，带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;

关闭线程池

/*
线程池状态变为 SHUTDOWN
- 不会接收新任务
- 但已提交任务会执行完
- 此方法不会阻塞调用线程的执行
*/
void shutdown();
/*
线程池状态变为 STOP
- 不会接收新任务
- 会将队列中的任务返回
- 并用 interrupt 的方式中断正在执行的任务
*/
List<Runnable> shutdownNow();

任务调度线程池

用来调度任务的执行时间

ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
// 添加两个任务，希望它们都在 1s 后执行
        System.out.println("程序开始执行时间"+new Date());
        executor.schedule(() -> {
            System.out.println("任务1，执行时间：" + new Date());
            try {
     Thread.sleep(2000); } catch (InterruptedException e) {
     }
        }, 1000, TimeUnit.MILLISECONDS);
        executor.schedule(() -> {
            System.out.println("任务2，执行时间：" + new Date());
        }, 1000, TimeUnit.MILLISECONDS);

在这里插入图片描述
我们可以看到任务在延迟一秒后执行。

定时任务

实现一个每周四18点执行的任务。

public static void main(String[] args) {
//当前时间
        LocalDateTime now = LocalDateTime.now();
//获取周四时间
        LocalDateTime time = now.withHour(18).minusMinutes(0).minusSeconds(0).minusNanos(0).with(DayOfWeek.THURSDAY);
        long period = 1000*60*60*24*7;
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
        //防止当前时间已经过了这周要执行的时间
        if(now.compareTo(time)>0){
            time = time.plusWeeks(1);
        }
      long initailDelay =   Duration.between(now,time).toMillis();
        pool.scheduleAtFixedRate(()->{
            System.out.println("牛逼");
        },initailDelay,period, TimeUnit.MILLISECONDS);
    }