简单明了的Java线程池
线程池
线程池从功能上来看,就是一个任务管理器。在Java
中,Executor
接口是线程池的根接口,其中只包含一个方法:
Executor
void execute(Runnable command); // 执行任务
ExecutorService
继承了Executor
接口,提供了一些线程池的基础方法:
void shutdown(); // 关闭线程池(不接受新任务,但是旧任务会执行)
List<Runnable> shutdownNow(); // 关闭线程池,返回待执行任务
boolean isShutdown(); // 线程池是否会关闭
boolean isTerminated(); // 关闭之前所有任务是否被关闭。(必须先执行过shutdown)
....
再往下是两种线程池的实现:ThreadPoolExecutor
和ForkJoinPool
。ThreadPoolExecutor
中维护了一个BlockingQueue
阻塞队列保存所有的待执行任务,而ForkJoinPool
中每一个线程都有自己的BlockingQueue
用来存储任务。
ThreadPoolExecutor
在ThreadPoolExecutor
的构造方法中,需要提供几个参数:corePoolSize
、maximumPoolSize
、keepAliveTime
、BlockingQueue
、RejectedExecutionHandler
。其中corePoolSize
表示当前线程池维护几个线程,maximumPoolSize
表示允许的最大线程数。keepAliveTime
表示如果当前线程数在
corePoolSize
和maximumPoolSize
之间时,允许在多久时间内保持存活等待新任务。BlockingQueue
是保存任务的阻塞队列,RejectedExecutionHandler
是不同的拒绝策略。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
那么
ThreadPoolExecutor
中是如何创建新线程呢?
在接到请求执行一个新任务时,首先会判断当前线程数是否大于corePoolSize
,如果没有则创建新线程。否则将当前任务放到阻塞队列中,如果当前队列已满,则创建新的线程执行任务。在成功将当前任务放到队列中之后,我们还需要二次判断当前线程池中是否有线程已经销毁或者当前线程池停止运行。当线程数量大于maximumPoolSize
时,执行拒绝策略。
这段代码在ThreadPoolExecutor
的excute
方法中体现,其中也有详细地对执行任务顺序的描述。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
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);
}
else if (!addWorker(command, false))
reject(command);
}
如何设置线程池的线程数量:如果是CPU密集型应用,则线程大小为N或者N+1;如果是IO密集型应用,则线程大小为2N或者2N+2
ForkJoinPool
ForkJoinPool
和ThreadPoolExecutor
都是继承自AbstractExecutorService
抽象类,所以它和ThreadPoolExecutor
的使用几乎没有多少区别,除了任务变成了ForkJoinTask
以外。
ForkJoinPool
和ThreadPoolExecutor
最主要的区别就是ForkJoinPool
中每一个线程都有属于自己的队列,当某个线程队列任务全部执行完了时,会通过"窃取工作"从别的线程队列中取出一个任务进行执行。
具体的策略就是每一个线程维护一个自己的队列,先进后出(FILO
)将任务塞到队列的头部,执行任务时从队列头部取出任务执行。其他线程从队列尾部窃取任务执行。减少阻塞消耗,特别适用于计算型任务。
Callable和Future
在jvm内存模型中我们会发现,每个线程有自己的内存空间,而且线程的run
方法返回空值。这个时候就会出现一个问题,如果想要在其他线程中拿到当前线程运行的结果是不可能的。所以就有了Callable
和Future
的存在,Callable
可以让线程返回值,而Future
可以拿到线程的返回值。
这里有一个误区,Callable
的使用不是在线程的run
方法中将返回值传递给call
方法,再从future.get()
中取值。实际上Callable
和Future
的配合使用,是利用了一个叫做FutureTask
的类,这个类同时继承了Runnable
和Future
接口,初始化的构造函数中会接受一个Callable
对象或者将Runnable
对象封装到Callable
对象中。在他的run
方法中,会调用call
方法。
public void run() {
if (state != NEW ||
!RUNNER.compareAndSet(this, null, Thread.currentThread()))
return;
try {
Callable<V> c = callable; // 获取传入的callable对象
if (c != null && state == NEW) {
V result; // call返回的结果
boolean ran;
try {
result = c.call(); // 执行call方法中的任务
ran = true; // 成功执行
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result); // 返回结果值
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
实际上这个set(result)
会将返回的结果塞到private Object outcome
对象中,这个就是我们最终通过future.get()
获取到的值。
RejectedExecutionHandler
上述线程池中我们提到拒绝策略,就是在线程池满时拒绝添加新线程,执行拒绝策略,就是通过这个RejectedExecutionHandler
进行处理。在ThreadPoolExecutor
线程池中,定义了4种不同的拒绝策略,他们都继承了RejectedExecutionHandler
。
- AbortPolicy
这个是默认的拒绝策略,他将丢弃当前的任务,并抛出异常。
/**
* A handler for rejected tasks that throws a
* {@link RejectedExecutionException}.
*
* This is the default handler for {@link ThreadPoolExecutor} and
* {@link ScheduledThreadPoolExecutor}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
他对当前的任务r
没有做任何处理。这是默认的拒绝策略,没有固定的使用场景,但是有一点需要注意,Executors
中提供的几种线程池的队列都是无界的,所以不会触发拒绝策略。
- DiscardPolicy
丢弃当前任务,不抛出异常。
/**
* A handler for rejected tasks that silently discards the
* rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
他既没有执行任务,也没有抛出异常。这种拒绝策略一般是你的任务无关紧要时使用,因为他不会返回异常。
- DiscardOldestPolicy
将当前队列头(即将被执行的任务)丢弃,执行当前任务。
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
* is shut down, in which case the task is discarded.
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
在这个拒绝策略中,他将当前队列头丢弃,调用e.execute(r)
重新尝试执行当前任务,当然他也会悄无声息的丢弃任务。这种一般是新任务的优先度更高时使用,比如说新消息来了,那么旧消息就无关紧要了。
- CallerRunsPolicy
在当前线程下运行该任务。
/**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
在当前线程执行该任务,相当于做了一个阻塞,当前线程暂停接受新任务并且当前线程不处于空闲状态,但是如果该任务执行时间过长可能会导致其他线程等待。这种一般是在不允许失败的、并发量较小的情况下使用。
工具类
Executors
Executors
是线程池工具类和工厂类,最主要的是提供了几种常见的线程池的创建。jdk8
以前一共是4种,而在1.8之后添加了forkjoin
连接池。
newSingleThreadExecutor
:这是单例的线程池,只包含一条线程,如果该线程意外中断,会创建新线程。newFixedThreadPool
:固定数量线程池,由于是固定大小的线程池,所以maximumPoolSize是没有意义的,只会创建corePoolSize个线程。newCachedThreadPool
:缓存线程池。corePoolSize
大小为0,maximumPoolSize
为Integer.MAX_VALUE
,60秒内无任务则销毁线程。ScheduledThreadPoolExecutor
:定时线程池。可以设置延时时间。
在jdk8
之后,Executors
添加了一种forkjoinpool
线程池:
newWorkStealingPool
: 工作窃取线程池,通过ForkJoinPool
实现。每个线程有独立的队列,完成之后从别的线程窃取工作。
当然除了除了上述之外,也可以通过ThreadPoolExecutor
或者ForkJoinPool
的构造函数来自定义线程池。
Semaphore
Semaphore
叫做信号量,我们可以通过设置个数来限制同时进入资源的线程数。比如一个停车场有十个停车位,我们设置停车位为10,那么同时只能有10辆车进入停车场。
Semaphore
的初始化过程中,我们可以提供两个参数,permits
和fair
分别表示信号量的个数和是否公平,如果fair
为true的话会使用公平锁的机制初始化线程队列同步器(AQS
)。公平锁和非公平锁的区别在于公平锁线程等待的时间越长越优先,所以公平锁的吞吐量比非公平锁小,非公平锁可能会造成某个线程等待时间过长。
在Semaphore
两种锁的具体实现就是在公平锁的tryAcquireShared
方法中多了一句:if (hasQueuedPredecessors()) return -1;
如果当前线程不是排在最前面的话,就返回-1;
Semaphore
使用acquire
方法来获取信号,通过release
方法在结束后释放信号。他的具体实现就是通过AQS
中的state
状态来保存promits
数量,被拿走就减去指定的数量。
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors()) // 判断当前线程是否在队列最前面
return -1;
int available = getState(); // 获取AQS中保存的promits数量
int remaining = available - acquires; // 总promits减去当前要取的信号量
if (remaining < 0 ||
compareAndSetState(available, remaining)) // CAS(compare and swap)乐观锁的方式更换state值
return remaining; // 返回剩下的信号量个数
}
}
可能看到这里会对AQS的概念比较模糊,如果想要进一步了解AQS的话,可以直接跳到AQS的部分,因为接下来的东西很多都跟AQS有关。
CountDownLatch
CountDownLatch
一般叫做计数器,他的作用是挂起线程等待其他线程运行到计算器清0之后再继续运行。一般用于流程控制,等待前置线程执行。
他也是通过AQS
来进行实现的,构造函数中要求提供一个count
值赋值给state
,调用countDown
之后计数器减1,直到state值
为0时,才能获取锁,继续执行下面的任务。
public static void main(String[] args) throws InterruptedException, ExecutionException {
CountDownLatch countDownLatch = new CountDownLatch(2);
Runnable taskMain = () -> {
try {
countDownLatch.await(); // 挂起,等待AQS的state值为0时被唤醒,解锁继续执行
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("继续执行任务");
};
Runnable task1 = () -> {
countDownLatch.countDown(); // 将AQS的state值减1 (计数器减1)
System.out.println("前置任务1完成");
};
Runnable task2 = () -> {
countDownLatch.countDown(); // 将AQS的state值减1 (计数器减1)
System.out.println("前置任务2完成");
};
new Thread(taskMain).start();
new Thread(task1).start();
new Thread(task2).start();
}
用法也很简单,在前置线程中使用countDown
计数减1,在后续线程中使用await
等待锁释放。
锁
AQS
AbstractQueuedSynchronizer
是concurrent
包下非常重要的一个接口,ReentranLock
、Semaphore
、CountDownLatch
等工具类的底层都是通过AQS
来实现。
那么AQS
底层是怎么来进行实现的呢?不同于synchronized
直接作用于jvm
底层,AQS
定义了一个volatile
属性的变量state
来表示当前资源是否被锁。那么如何在多线程下对一个变量进行修改?这里借用了乐观锁的概念,采用CAS
的方式来修改变量。CAS
指的是compare and swap
,它假设所有线程对资源的访问是没有冲突的,如果有冲突,则通过比较交换的方式来解决。如果要修改,则会传两个值,一个是state
的预期值,一个是修改之后的值,如果预期值跟state
值一致,那么就允许更改,否则则判断是有别的线程已经对state
修改过了,表示锁定状态。这个时候当前线程会被挂起,放到CLH队列中,等待其他线程释放state
唤醒。
最新文章
- CentOS6.8 修改主机名(1)
- How To Handle a Loss of Confidence in Yourself
- NLS_LANG
- jquery选择器效率优化问题
- cadence遇到的问题(持续更新)
- Python学习笔记——基础篇【第一周】——变量与赋值、用户交互、条件判断、循环控制、数据类型、文本操作
- python3.6安装pyspider
- jq 时间计算
- python开发concurent.furtrue模块:concurent.furtrue的多进程与多线程&;协程
- Struts 2 之配置文件
- 【原创】大数据基础之Benchmark(4)TPC-DS测试结果(hive/hive on spark/spark sql/impala/presto)
- mkpasswd命令
- ArcGIS栅格影像怎么从WGS84地理坐标转成Xian80投影坐标
- 【读书笔记】iOS-网络-Web Service协议与风格
- Linux学习笔记:常用命令grep、iconv、cp、mv、rm
- Servlet 3.0 新特性详解
- [html]window.open 使用示例
- NHibernate 3 Beginner&#39;s Guide
- spring boot 总结
- Ubuntu 手机 app开发学习0
热门文章
- Python3.9安装PySpider步骤及问题解决
- Orchestrator+Proxysql 实现自动导换+应用透明读写分离
- C# MongoDB添加索引
- 【算法学习笔记】动态规划与数据结构的结合,在树上做DP
- 自学linux——2.认识目录及常用指(命)令
- 手写Pascal解释器(二)
- Linux下MySQL基础及操作语法
- Commons-Beanutils利用链分析
- Ming Yin(@kalasoo)在知乎的几个回答 : 观点犀利
- 如何看待Android开发的“前景和内卷”