JUC-线程池

骑猪看日落 2024-04-24 14:49 6阅读 0赞

## ThreadPoolExecutor ##

### 线程状态 ###

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

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

这些信息存储在原子变量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)

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

*  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<Integer> 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();

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

### 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);

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

### 定时任务 ###

实现一个每周四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);
        }

[7b60807fd5834720a27bb9a80edcebde.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/24/946eb4362755434093ec73aeb7961a5c.png
[2900b48317a2495ebcdb74b04917271a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/24/b301f69f8e31466a9ddb0038810f015f.png
[e43c02680391410da78c1946d9b1cecd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/24/2aaf51ae2d1742e7a1388f88a247a7c3.png