Java多线程 4.线程池

Posted 2020-12-03 kepus

1.Java多线程-认识Java线程

2.Java多线程-线程安全

3.Java多线程-线程协作

4.Java多线程-线程池

5.Java多线程-栅栏

6.Java多线程-Fork/Join

 

---

4.1 线程池介绍

 4.1.1 什么是线程池

　　类比数据库连接池，线程池就是放线程的池子，当程序有任务需要线程执行时，程序可以将任务提交给线程池，线程池会根据线程池的配置来处理提交的任务；

处理情况有 1.创建新的线程执行任务；2.将任务放到任务队列，等待空闲线程执行任务； 3.拒绝任务.

 4.1.2 重要接口和类

　　 JDK的线程池实现类主要有普通线程池(ThreadPoolExecutor)和具有调度功能的线程池(ScheduledThreadPoolExecutor)，可以通过new新建的方式创建以上两种线程池，也可以使用JDK提供的Executors工具类来创建。

 

Runnable/Callable: 程序定义任务的接口，Runnable任务没有返回值，Callable任务可以有返回值；

1. Runnable接口实现run方法
2. Callable接口实现call方法且有返回值

ThreadPoolExecutor：线程池核心类

1. 属性：任务队列BlockingQueue<Runnable>， 存放线程池中的任务
2. 构造方法：ThreadPoolExecutor(corePoolSize, maximumPoolSize,  keepAliveTime, TimeUnit, BlockingQueue<Runnable>, ThreadFactory, RejectedExecutionHandler)
   
3. 执行任务：execute(Runnable)：提交Runnable任务， 任务直接进入线程池的任务队列
4. 提交任务：submit(Runnable, T)：返回Future<T>, Runnable和T分别赋值给RunnableAdapter的task和result， 然后RunnableAdapter复制给FutureTask，FutureTask进入线程池的任务队列
5. 提交任务：submit(Callable<T>)：返回Future<T>，Callable赋值给FutureTask的task，FutureTask进入线程池的任务队列， Callable的call方法返回值赋值给FutureTask的result

Future: 获取线程池中任务执行情况的接口

1. get(): 阻塞获取任务执行结果
2. get(timeout, TimeUnit): 设定阻塞时间，获取任务执行结果
3. isDone(): 非阻塞查看任务是否执行结束

FutureTask：任务和任务结果的封装类，submit方法提交的任务最终都封装成FutureTask，然后放入线程池的任务队列

1. Callable: 任务
2. Object: 任务的执行结果

RunnableAdapter: submit(Runnable)方法提交的任务不能直接封装成FutureTask， 需要将Runnable复制给RunnableAdapter的task，然后复制给FutureTask的Callable

1. 属性Runnable：submit提交的Runnable任务
2. 属性T：submit提交的Runnable任务的结果（注意Runnable的执行结果时任务提交时已经确定），最终赋值给FutureTask的outcom

ScheduledThreadPoolExecutor: 具有调度能力的线程池

1. schedule:一定时间后执行任务
2. scheduleAtFixedRate: 周期性的执行任务，无论上一次的任务是否执行结束，都会周期性开始新的任务
3. scheduleWithFixedDelay：周期性的执行任务，从上一次任务执行结束开始间隔一定时间执行新的任务

 4.1.3 线程池使用

线程池中线程创建过程

  线程池的配置中有corePoolSize、maximumPoolSize、BlockingQueue 三个属性，分别表示线程池的核心线程数、最大线程数、任务队列，当向线程池提交一个任务时会有以下情况：

  1.池中线程数 小于 corePoolSize： 新建线程执行新提交的任务

  2.池中线程数 大于 corePoolSize ：任务放入任务队列

　 2.1 任务队列未满：等待空闲线程执行任务

    2.2 任务队列已满:

　  2.2.1 池中线程数小于 maximumPoolSize： 新建线程执行任务（注意是执行新提交的任务，而不是在任务队列取久的任务）

　  2.2.2 池中线程数大于 maximumPoolSize: 则用RejectedExecutionHandler处理任务（默认抛出异常）

线程池工具类：JDK提供了Executors工具类来创建普通线程池和调度线程池

 >创建普通线程池

 1 import java.util.concurrent.Callable;
 2 import java.util.concurrent.Executors;
 3 import java.util.concurrent.Future;
 4 import java.util.concurrent.ThreadPoolExecutor;
 5 import java.util.concurrent.TimeUnit;
 6 
 7 public class ThreadPool {
 8 
 9     public static void main(String[] args) throws Exception {
10 
11         //通过Executors的静态方法创建线程池
12         ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
13 
14         //提交任务到线程池，并得到Future
15         Future<String> future = threadPoolExecutor.submit(new Callable<String>() {
16             @Override
17             public String call() throws Exception {
18                 TimeUnit.SECONDS.sleep(10);
19                 return "任务执行了10秒钟，结束了.";
20             }
21         });
22 
23         //阻塞等待任务执行结束，并打印任务执行结果
24         System.out.println(future.get());
25     }
26 }

View Code

 >创建调度线程池

 1 import java.time.Instant;
 2 import java.time.ZoneId;
 3 import java.time.format.DateTimeFormatter;
 4 import java.util.concurrent.Executors;
 5 import java.util.concurrent.ScheduledExecutorService;
 6 import java.util.concurrent.TimeUnit;
 7 
 8 public class ThreadPool {
 9 
10     private static DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss").withZone(ZoneId
11             .systemDefault());
12 
13     public static void main(String[] args) throws Exception {
14         //通过Executors的静态方法创建线程池
15         ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
16         //每隔5秒执行一次
17         scheduledExecutorService.scheduleAtFixedRate(() -> {
18             System.out.println("周期执行的任务：" + dateTimeFormatter.format(Instant.now()));
19         }, 0, 5, TimeUnit.SECONDS);
20     }
21 }

View Code

4.2 线程池扩展

　　在java中对一个类的扩展常有2中情况：重写类的protect方法和传入public方法中具有回调特性的可选参数，而ThreadPoolExecutor是线程池的核心类，所以线程池的扩展需要围绕此类展开

 4.2.1 重写保护方法：ThreadPoolExecutor中有好多protect方法，在此以扩展beforeExecute和afterExecute

　　beforeExecute(Thread t, Runnable r)：任务执行前调用的方法， t为执行任务的线程，r为待执行的任务，r可能为实现Runnable接口的任务（execute方式提交），也可能是FutureTask任务（submit方式提交)

　　如果r为FutureTask那么FutureTask的Callable属性可能是实现了Callable的任务类（submit(Callable)），可能是RunnableAdapter类(submit(Runnable))

　　afterExecute(Runnable r, Throwable t)：任务执行结束调用的方法，r同beforeExecute， t为线程执行过程抛出的异常

 4.2.2 传入构造方法中的可选参数：ThreadPoolExecutor有4个构造参数

　　最少参数构造器：ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)

　　最多参数构造器：ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 

　　通过比较发现有2个参数为可选参数ThreadFactory 和 RejectedExecutionHandler， ThreadFactory为创建线程池中线程的工厂，RejectedExecutionHandler为线程池拒绝任务时的处理类

　　多线程开发的第一关注点就是给线程起一个好的名字，所以扩展线程池一般扩展ThreadFactory

 4.2.3 一般通过自定义一个Executors来扩展线程池

 1 import java.util.concurrent.FutureTask;
 2 import java.util.concurrent.SynchronousQueue;
 3 import java.util.concurrent.ThreadFactory;
 4 import java.util.concurrent.ThreadPoolExecutor;
 5 import java.util.concurrent.TimeUnit;
 6 import java.util.concurrent.atomic.AtomicInteger;
 7 
 8 /**
 9  * 线程池扩展类
10  */
11 public class GCExecutors {
12 
13     public static void main(String[] args) throws Exception {
14 
15         // 创建线程池
16         ThreadPoolExecutor threadPoolExecutor = newThreadPoolExecutor(2, 2, 100, "线程");
17 
18         //向线程池提交任务
19         threadPoolExecutor.execute(() -> {
20             try {
21                 TimeUnit.SECONDS.sleep(5);
22             } catch (InterruptedException e) {
23                 e.printStackTrace();
24             }
25         });
26 
27         threadPoolExecutor.submit(() -> {
28         });
29     }
30 
31     /**
32      *
33      * @param corePooSize 线程池中核心线程数
34      * @param maxnumPoolSize 线程池中最大线程数
35      * @param keepAliveTime 超过corePoolSize的线程的空闲时间
36      * @param threadFlag 线程池中线程的名字的标识
37      * @return 线程池
38      */
39     public static ThreadPoolExecutor newThreadPoolExecutor(int corePooSize, int maxnumPoolSize, long keepAliveTime,
40                                                            String threadFlag) {
41 
42         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePooSize, maxnumPoolSize, keepAliveTime,
43                 TimeUnit.SECONDS, new SynchronousQueue(), new GCThreadFactory(threadFlag)) {
44 
45             /**
46              * 任务执行前执行的方法
47              * @param t 执行任务的线程
48              * @param r 待执行的任务
49              */
50             @Override
51             protected void beforeExecute(Thread t, Runnable r) {
52                 System.out.println(t.getName() + " will execute new task.");
53                 if (r instanceof FutureTask) {
54                     // 可以通过反射取出callable,然后通过instance of判断callable是不是你的Callable任务实现类
55                     System.out.println("任务是通过submit提交的，但是不能区分是Runnable还是Callable");
56                 } else {
57                     System.out.println("任务是通过execute提交的");
58                 }
59             }
60 
61             /**
62              * 任务执行结束执行的方法
63              * @param r 执行结束的任务
64              * @param t 任务执行过程抛出的异常
65              */
66             @Override
67             protected void afterExecute(Runnable r, Throwable t) {
68                 System.out.println("任务执行结束了");
69             }
70         };
71         return threadPoolExecutor;
72     }
73 }
74 
75 /**
76  * 自定义线程工厂
77  */
78 class GCThreadFactory implements ThreadFactory {
79 
80     private String threadFlag;
81 
82     private AtomicInteger threadNum = new AtomicInteger();
83 
84     public GCThreadFactory(String threadFlag) {
85         this.threadFlag = threadFlag;
86     }
87 
88     @Override
89     public Thread newThread(Runnable runnable) {
90         return new Thread(runnable, String.format("pool-%s-%s", threadFlag, threadNum.getAndIncrement()));
91     }
92 }