- 1用Processing写一个根据音乐变化的黑白格游戏
- 2github javaguide_GitHub上收获Star数排名前10的Java项目
- 3gepc 增强数据的加载 patch enhance_.stack(patches)
- 4（一）手把手教你如何通过ARM DesignStart计划在FPGA上搭建一个Cortex-M3软核_手把手教你搭建cortex软核
- 5pixhawk 光流--位置估计--姿态估计--位置控制--姿态控制_motorspeed = map(abs(tiltangle), 0, 90, 0, 500);
- 6数据结构之二叉树（JAVA实现）_java实现二叉树
- 7yolov5 配置环境，训练自己的数据集，实现识别图片，视频，视频流及代码注释_yolov5检测视频流
- 8select case when if 的一些用法
- 9解析xml文件的几种技术_解释xml文档内容的技术包括下列哪些项
- 10基于python的人脸识别门禁系统_python人脸识别门禁系统
java——线程池_java线程池
赞
踩
一、线程池
线程池可以看做是线程的集合。它的工作主要是控制运行的线程的数量，处理过程中将任务放入队列，然后在线程创建后 启动这些任务，如果线程数量超过了最大数量超出数量的线程排队等候，等其它线程执行完毕， 再从队列中取出任务来执行。他的主要特点为：线程复用；控制最大并发数；管理线程。
线程复用：
每一个 Thread 的类都有一个 start 方法。 当调用 start 启动线程时 Java 虚拟机会调用该类的 run 方法。 那么该类的 run() 方法中就是调用了 Runnable 对象的 run() 方法。 我们可以继承重写 Thread 类，在其 start 方法中添加不断循环调用传递过来的 Runnable 对象。 这就是线程池的实 现原理。循环方法中不断获取 Runnable 是用 Queue 实现的，在获取下一个 Runnable 之前可以 是阻塞的。
一般的线程池主要分为以下 4 个组成部分：
- 线程池管理器：用于创建并管理线程池
- 工作线程：线程池中的线程
- 任务接口：每个任务必须实现的接口，用于工作线程调度其运行
- 任务队列：用于存放待处理的任务，提供一种缓冲机制
Java 中的线程池是通过 Executor 框架实现的，该框架中用到了 Executor，Executors， ExecutorService，ThreadPoolExecutor ，Callable 和 Future、FutureTask 这几个类。
线程池工作过程 ：
- 线程池刚创建时，里面没有一个线程。任务队列是作为参数传进来的。不过，就算队列里面
有任务，线程池也不会马上执行它们。 - 当调用 execute() 方法添加一个任务时，线程池会做如下判断：
a) 如果正在运行的线程数量小于 corePoolSize，那么马上创建线程运行这个任务；
b) 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入队列；
c) 如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要
创建非核心线程立刻运行这个任务；
d) 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池
会抛出异常 RejectExecutionException。 - 当一个线程完成任务时，它会从队列中取下一个任务来执行。
- 当一个线程无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运
行的线程数大于 corePoolSize，那么这个线程就被停掉。所以线程池的所有任务完成后，它
最终会收缩到 corePoolSize 的大小。
创建和销毁线程所花费的时间和资源可能比处理任务所花费的时间和资源还要多，所以说：我们的线程最好是交由线程池来管理，这样可以减少对线程生命周期的管理，一定程度上提高性能。
二、JDK提供的线程池API
DK给我们提供了Excutor框架来使用线程池，它是线程池的基础。
- Executor提供了一种将**“任务提交”与“任务执行”**分离开来的机制(解耦)
Executor接口：
public interface Executor {
/**
* 执行任务
*/
void execute(Runnable command);
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
ExcutorService接口：
public interface ExecutorService extends Executor { /** * 关闭线程池 */ void shutdown(); /** * 停止所有主动执行的任务以及停止等待任务的处理，并返回任务列表 */ List<Runnable> shutdownNow(); /** * 是否关闭线程池 */ boolean isShutdown(); /** * 判断是否所有任务在关闭后全部完成 */ boolean isTerminated(); /** * 判断任务执行超时还是提前终止 * 超时:true * 提前终止:false */ boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; /** * 提交任务用于执行 */ <T> Future<T> submit(Callable<T> task); <T> Future<T> submit(Runnable task, T result); Future<?> submit(Runnable task); /** * 执行给定的任务 */ <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException; <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException; <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
AbstractExecutorService类：
对于ExecutorService接口的默认实现，同时新增了三个方法。
public abstract class AbstractExecutorService implements ExecutorService { protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return new FutureTask<T>(runnable, value); } protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); } private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException { if (tasks == null) throw new NullPointerException(); int ntasks = tasks.size(); if (ntasks == 0) throw new IllegalArgumentException(); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks); ExecutorCompletionService<T> ecs = new ExecutorCompletionService<T>(this); // For efficiency, especially in executors with limited // parallelism, check to see if previously submitted tasks are // done before submitting more of them. This interleaving // plus the exception mechanics account for messiness of main // loop. try { // Record exceptions so that if we fail to obtain any // result, we can throw the last exception we got. ExecutionException ee = null; final long deadline = timed ? System.nanoTime() + nanos : 0L; Iterator<? extends Callable<T>> it = tasks.iterator(); // Start one task for sure; the rest incrementally futures.add(ecs.submit(it.next())); --ntasks; int active = 1; for (;;) { Future<T> f = ecs.poll(); if (f == null) { if (ntasks > 0) { --ntasks; futures.add(ecs.submit(it.next())); ++active; } else if (active == 0) break; else if (timed) { f = ecs.poll(nanos, TimeUnit.NANOSECONDS); if (f == null) throw new TimeoutException(); nanos = deadline - System.nanoTime(); } else f = ecs.take(); } if (f != null) { --active; try { return f.get(); } catch (ExecutionException eex) { ee = eex; } catch (RuntimeException rex) { ee = new ExecutionException(rex); } } } if (ee == null) ee = new ExecutionException(); throw ee; } finally { for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } } ...... }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
ScheduledExecutorService接口：
ThreadPoolExecutor类：
这个类是我们用的最多的类，要重点掌握！！！
ScheduledThreadPoolExecutor类：
相当于提供了“延迟”和“周期性”执行功能的ThreadPoolExecutor类
2.1 Callable和Future
我们都知道创建线程的两种方式，一是继承Thread类，重写run方法、二是实现Runable接口，重写run方法。
public class TestThread extends Thread{ @Override public void run() { } } ---------------- public class TestThread implements Runnable{ @Override public void run() { } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
可以看到run()方法是没有返回值的。那我们想要执行多线程任务后返回给我们任务的结果怎么办?
让我们看看第三种创建线程的方式：实现Callabl接口。重写call方法。
通过上边线程池的简单了解，我们发现：很多的API都有Callable和Future这么两个东西。
Future<?> submit(Runnable task)
<T> Future<T> submit(Callable<T> task)
- 1
- 2
简单看一下Callable接口，这是个函数式接口，里面只有一个call方法，注释很简单：计算结果，如果无法计算则抛出异常。return的是计算的结果。
当我们的任务需要返回值的时，我们就可以使用Callable！Future和Callable通常是一起使用的，
它一般我们认为是Callable的返回值，但他其实代表的是任务的生命周期(它是能获取得到Callable的返回值的)。我们看下Future接口的源码：
public interface Future<V> { /** * 用于停止任务。如果尚未启动，它将停止任务。如果已启动，则仅在mayInterrupt为true时才会中断任务。 */ boolean cancel(boolean mayInterruptIfRunning); /** * 判断任务是否被取消 * 如果此任务在完成之前已取消则为true */ boolean isCancelled(); /** * 如果任务完成，则返回true，否则返回false */ boolean isDone(); /** * 用于获取任务的结果。如果任务完成，它将立即返回结果，否则将等待任务完成，然后返回结果。 */ V get() throws InterruptedException, ExecutionException; V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
Future和Callable示例：
public class Test { public static void main(String[] args) throws ExecutionException, InterruptedException { FutureTask<Integer> future = new FutureTask<>(new CallableDemo(100)); Thread thread = new Thread(future); thread.start(); //获取任务执行返回值 Integer result = future.get(); System.out.println(result); } } class CallableDemo implements Callable<Integer>{ private int number; public CallableDemo(int number) { this.number = number; } @Override public Integer call() { int sum = 0; for (int x = 1; x <= number; x++) { sum += x; } return sum; } } console: 5050
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
FutureTask实现了RunnableFuture接口，而RunnableFuture继承了Runnable和Future，也就是说FutureTask既是Runnable，也是Future。
三、ThreadPoolExecutor详解
ThreadPoolExecutor在构造方法：
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
构造函数的参数含义如下：
- corePoolSize： 是指线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到workQueue任务队列中去，简单理解就是corePoolSize表示允许线程池中允许同时运行的最大线程数；
- maximumPoolSize： 线程池中最大线程数量**，**这个参数会根据你使用的workQueue任务队列的类型，决定线程池会开辟的最大线程数量；maximumPoolSize肯定是大于等于corePoolSize
- keepAliveTime： 当线程池中空闲线程数量超过corePoolSize时，多余的线程会在多长时间内被销毁；
- unit： keepAliveTime的单位；
- workQueue： 任务队列，被添加到线程池中，但尚未被执行的任务；它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列几种；
- threadFactory： 线程工厂，用于创建线程，一般用默认即可；
- handler： 任务拒绝策略，当任务太多来不及处理时，如何拒绝任务；
线程数量要点：
- 如果运行线程的数量少于核心线程数量，则创建新的线程处理请求
- 如果运行线程的数量大于核心线程数量，小于最大线程数量，则当队列满的时候才创建新的线程
- 如果核心线程数量等于最大线程数量，那么将创建固定大小的连接池
- 如果设置了最大线程数量为无穷，那么允许线程池适合任意的并发数量
线程空闲时间要点：
- 当前线程数大于核心线程数，如果空闲时间已经超过了，那该线程会销毁。
排队策略要点：
- 同步移交：不会放到队列中，而是等待线程执行它。如果当前线程没有执行，很可能会新开一个线程执行。
- 无界限策略：如果核心线程都在工作，该线程会放到队列中。所以线程数不会超过核心线程数
- 有界限策略：可以避免资源耗尽，但是一定程度上减低了吞吐量
当线程关闭或者线程数量满了和队列饱和了，就有拒绝任务的情况了：
拒绝任务策略：
- 直接抛出异常
- 使用调用者的线程来处理
- 直接丢掉这个任务
- 丢掉最老的任务
3.1 workQueue任务队列
workQueue任务队列一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列；
1、直接提交队列synchronousQueue： 这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务
2、有界任务队列ArrayBlockingQueue： 基于数组的先进先出队列，此队列创建时必须指定大小；
3、无界任务队列LinkedBlockingQueue： 基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE；
4、优先任务队列PriorityBlockingQueue： 优先任务队列通过PriorityBlockingQueue实现
直接提交队列示例：
public class ThreadPoolTest {
private static ExecutorService pool;
public static void main(String[] args) {
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS,new SynchronousQueue<Runnable>(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
for (int i = 0; i < 3; i++) {
pool.execute(new ThreadDemo());
}
}
}
class ThreadDemo implements Runnable{
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.zsn.scheduler.Thread.ThreadDemo@1be6f5c3 rejected from java.util.concurrent.ThreadPoolExecutor@6b884d57[Running, pool size = 2, active threads = 2, queued tasks = 0, completed tasks = 0] at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063) at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379) at com.zsn.scheduler.Thread.ThreadPoolTest.main(ThreadPoolTest.java:20)pool-1-thread-1pool-1-thread-2
- 1
使用SynchronousQueue队列，提交的任务不会被保存，总是会马上提交执行。如果用于执行任务的线程数量小于maximumPoolSize，则尝试创建新的进程，如果达到maximumPoolSize设置的最大值，则根据你设置的handler执行拒绝策略
有界任务队列示例：
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
- 1
使用ArrayBlockingQueue有界任务队列，若有新的任务需要执行时，线程池会创建新的线程，直到创建的线程数量达到corePoolSize时，则会将新的任务加入到等待队列中。若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数量达到maximumPoolSize设置的最大线程数量，若大于maximumPoolSize，则执行拒绝策略。
无界任务队列示例：
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
- 1
使用无界任务队列，线程池的任务队列可以无限制的添加新的任务，而线程池创建的最大线程数量就是你corePoolSize设置的数量，也就是说在这种情况下maximumPoolSize这个参数是无效的，哪怕你的任务队列中缓存了很多未执行的任务，当线程池的线程数达到corePoolSize后，就不会再增加了；若后续有新的任务加入，则直接进入队列等待，当使用这种任务队列模式时，一定要注意你任务提交与处理之间的协调与控制，不然会出现队列中的任务由于无法及时处理导致一直增长，直到最后资源耗尽的问题。
优先任务队列PriorityBlockingQueue：
PriorityBlockingQueue它其实是一个特殊的无界队列，它其中无论添加了多少个任务，线程池创建的线程数也不会超过corePoolSize的数量，只不过其他队列一般是按照先进先出的规则处理任务，而PriorityBlockingQueue队列可以自定义规则根据任务的优先级顺序先后执行。
3.2 拒绝策略
线程池中的线程已经用完了，无法继续为新任务服务，同时，等待队列也已经排满了，再也
塞不下新任务了。这时候我们就需要拒绝策略机制合理的处理这个问题。
JDK 内置的拒绝策略如下：
- AbortPolicy ： 直接抛出异常，阻止系统正常运行。
- CallerRunsPolicy ： 只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的
任务。显然这样做不会真的丢弃任务，但是，任务提交线程的性能极有可能会急剧下降。 - DiscardOldestPolicy ： 丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再
次提交当前任务。 - DiscardPolicy ： 该策略默默地丢弃无法处理的任务，不予任何处理。如果允许任务丢
失，这是最好的一种方案。
以上内置拒绝策略均实现了 RejectedExecutionHandler 接口，若以上策略仍无法满足实际
需要，完全可以自己扩展 RejectedExecutionHandler 接口。
3.3 已默认实现的池
下面我就列举三个比较常见的实现池：
- newFixedThreadPool
- newCachedThreadPool
- SingleThreadExecutor
newFixedThreadPool
newFixedThreadPool是一个固定线程数的线程池，它将返回一个corePoolSize和maximumPoolSize相等的线程池。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
}
- 1
- 2
- 3
newCachedThreadPool
创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行 很多短期异步任务的程序而言，这些线程池通常可提高程序性能。调用 execute 将重用以前构造 的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并 从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}
- 1
- 2
- 3
newScheduledThreadPool
创建一个线程池，它可安排在给定延迟后运行命令或者定期地执行
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
- 1
- 2
- 3
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
scheduledThreadPool.schedule(newRunnable(){
@Override
public void run() {
System.out.println("延迟三秒");
}
}, 3, TimeUnit.SECONDS);
scheduledThreadPool.scheduleAtFixedRate(newRunnable(){
@Override
public void run() {
System.out.println("延迟 1 秒后每三秒执行一次");
}
},1,3,TimeUnit.SECONDS);
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
newSingleThreadExecutor
Executors.newSingleThreadExecutor()返回一个线程池（这个线程池只有一个线程）,这个线程池可以在线程死后（或发生异常时）重新启动一个线程来替代原来的线程继续执行下去！
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()));}
- 1
- 2
四、execute执行方法
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); //如果线程池中运行的线程数量<corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //如果线程池中运行的线程数量>=corePoolSize，且线程池处于RUNNING状态，且把提交的任务成功放入阻塞队列中，就再次检查线程池的状态， // 1.如果线程池不是RUNNING状态，且成功从阻塞队列中删除任务，则该任务由当前 RejectedExecutionHandler 处理。 // 2.否则如果线程池中运行的线程数量为0，则通过addWorker(null, false)尝试新建一个线程，新建线程对应的任务为null。 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } // 如果以上两种case不成立，即没能将任务成功放入阻塞队列中，且addWoker新建线程失败，则该任务由当前 RejectedExecutionHandler 处理。 else if (!addWorker(command, false))reject(command); }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
五、线程池关闭
ThreadPoolExecutor提供了shutdown()和shutdownNow()两个方法来关闭线程池
- 调用shutdown()后，线程池状态立刻变为SHUTDOWN，而调用shutdownNow()，线程池状态立刻变为STOP。
- shutdown()等待任务执行完才中断线程，而shutdownNow()不等任务执行完就中断了线程。
