link-JAVA多线程与高并发系列[前言,大纲,目录]
目录
-
- 前置知识
-
- Executor:
- ExecutorService:
- Future&Callable
- FutureTask(比较常用)
- CompletableFuture(非常灵活)
- 认识ThreadPoolExecutor
-
- ThreadPoolExecutor的七个重要参数:
- 测试小例子:
- 调整线程池的大小
- ThreadPoolExecutor源码解析
-
- 1、常用变量的解释
- 2、构造方法
- 3、提交执行task的过程
- 4、addworker源码解析
- 5、线程池worker任务单元
- 6、核心线程执行逻辑-runworker
前置知识
关键类:
Executor <-extends- ExecutorService <-implements-AbstractExecutorService <-extends- ExecutorService
Callable,Future
Executor:
任务的定义和执行分开,只有一个执行Runnable方法:
void execute(Runnable command);
ExecutorService:
除了继承Executor可以执行任务的功能,还完善了整个任务执行器(线程池)的生命周期.比如,一个线程池里面有很多线程,怎么提交任务,执行完任务之后应该怎么处理线程,怎么关闭等等.
Future&Callable
可以看到ExecutorService里面有个方法,提交异步的任务:
Future submit(Callable task);
Callable:和Runnable类似,是一个任务,只不过它执行完后有返回值,有了返回值就可以有各种玩法了.
Future:Callable执行完后有一个返回值,通过Future可以拿到这个结果.
存储了一个会在将来产生的结果.
看个简单的例子:
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService service = Executors.newCachedThreadPool();
Future<String> future = service.submit(() -> {
// spend some seconds doing something
TimeUnit.SECONDS.sleep(2);
return "Hello World!";
}); //异步
System.out.println("i will get~");
System.out.println(future.get());//阻塞
service.shutdown();
}
FutureTask(比较常用)
FutureTask implements RunnableFuture,RunnableFuture extend Runnable, Future
FutureTask内部维护了一个Callable成员变量
前面用的Callable只是一个任务,Future只是一个返回值,这个FutureTask就是结合了一下,既是任务又是返回值.(Apple+pen->pineapple! 😃)
线程池WorkStealingPool和ForkJoinPool用到了FutureTask
看一个小例子:
public static void main(String[] args) throws InterruptedException, ExecutionException {
FutureTask<Integer> task = new FutureTask<>(() -> {
TimeUnit.MILLISECONDS.sleep(500);
return 1000;
}); //new Callable () { Integer call();}
// 可以是线程,也可以是线程池
new Thread(task).start();
System.out.println(task.get()); //阻塞
}
CompletableFuture(非常灵活)
CompletableFuture implements Future,CompletionStage
一个典型的应用场景:
有很多个子系统,他们各自有自己的数据库存储系统,可能是MySQL/Oracle/MongoDB等,现在需要统计他们的指标(比如平均请求响应时间)在一张大屏上展示分析.如果串行去查询子系统的数据,那这个分析的API就执行太久了,但是我们使用CompletableFuture,多线程异步执行,那时间就大大缩短.
当然这个场景用普通线程执行Callable也是可以搞定的,只是用CompletableFuture比较方便,相当于JDK已经造好轮子了,我们可以直接用它.
show my code:
public class TestCompletableFuture {
public static void main(String[] args) throws ExecutionException, InterruptedException {
normalTest();
futureTest();
// test002();
}
private static void normalTest() {
long start = System.currentTimeMillis();
Map<String, Double> metrics = new HashMap<>(4);
metrics.put("metricsOfMySQL", metricsOfMySQL());
metrics.put("metricsOfOracle", metricsOfOracle());
metrics.put("metricsOfMongoDB", metricsOfMongoDB());
long end = System.currentTimeMillis();
System.out.println("use serial method call! " + (end - start));
System.out.println(metrics + "\n-------------------------------------");
}
private static void futureTest() {
long start = System.currentTimeMillis();
Map<String, Double> metrics = new HashMap<>(4);
CompletableFuture<Double> metricsOfMySQL = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMySQL)
.thenApply(value -> metrics.put("metricsOfMySQL", value));
CompletableFuture<Double> metricsOfOracle = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfOracle)
.thenApply(value -> metrics.put("metricsOfOracle", value));
CompletableFuture<Double> metricsOfMongoDB = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMongoDB)
.thenApply(value -> metrics.put("metricsOfMongoDB", value));
System.out.println(metrics);
CompletableFuture.allOf(metricsOfMySQL, metricsOfOracle, metricsOfMongoDB).join();
long end = System.currentTimeMillis();
System.out.println("use completable future! " + (end - start));
System.out.println(metrics);
}
// 其他的用法,灵活的处理结果,有点函数式编程的感觉
private static void test002() {
// 异步执行
CompletableFuture.supplyAsync(() -> metricsOfMongoDB())
.thenApply(String::valueOf)
.thenApply(str -> "price " + str)
.thenAccept(System.out::println);
// 阻塞住主线程,等待上面执行完
try {
System.in.read();
} catch (IOException e) {
e.printStackTrace();
}
}
private static double metricsOfMySQL() {
sleepRandom();
return 2.00;
}
private static double metricsOfOracle() {
sleepRandom();
return 3.00;
}
private static double metricsOfMongoDB() {
sleepRandom();
return 1.00;
}
private static void sleepRandom() {
int time = new Random().nextInt(500);
try {
TimeUnit.MILLISECONDS.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("After %s sleep!\n", time);
}
}
}
认识ThreadPoolExecutor
最开始我们创建一个线程,执行一个Runnable任务,执行完后就销毁线程了.
然而创建一个线程需要跟操作系统申请资源,这个过程是比较耗时的;所以我们最好是让线程复用,即让一个线程去持续执行不同的任务,而不是执行一个任务后就销毁.(可以类比一下泡面桶和陶瓷碗)
线程池里面不仅仅是线程,它维护这两个集合,一个是线程集合,一个是任务集合.
ThreadPoolExecutor的七个重要参数:
-
int corePoolSize
核心线程数,最开始线程池里面没有线程,来了任务后会先创建一定数量的核心线程去执行任务;一般没有任务执行时也不会回收核心线程.
-
int maximumPoolSize
最大线程数.当任务比较多,核心线程执行不过来时会放入任务队列,任务队列满了后会创建非核心线程,maximumPoolSize=临时线程数+核心线程数,主要负责控制临时线程数.非核心线程在空闲一段时间后会被回收.
-
long keepAliveTime
生存时间.当一个非核心线程很长时间不执行任务了,就销毁该线程,这个参数就是控制空闲阈值.
核心线程默认不受此控制,也可以设置参数指定核心线程受此控制(allowCoreThreadTimeOut).
-
TimeUnit unit
生存时间单位,见名知意
-
BlockingQueue workQueue
任务队列
-
ThreadFactory threadFactory
线程工厂,自定义创建线程的方式.
有个默认的DefaultThreadFactory,指定了线程名字,daemon=false,priority=5(NORM_PRIORITY).不要小看线程名,多线程环境追踪错误日志时大有用处.
-
RejectedExecutionHandler handler
拒绝策略,当任务很多,任务队列满了,非核心线程数也达到上限后,再来任务的时候的处理策略.拒绝策略可以自定义,JDK提供了四种拒绝策略:
- AbortPolicy:抛异常,这也是默认的拒绝策略
- DiscardPolicy:安静的丢掉
-
DiscardOldestPolicy:丢掉队列中最老的任务,把新的放入队列
做游戏的时候可能会用,比如一个角色的每次移动作为一个操作当如线程池中,正常情况是依次移动;当队列满了就把最老的丢掉,减少影响.
-
CallerRunsPolicy:提交任务者(调用execute的线程)处理该任务
(实战中这四种一般都不用,而是自定义)
阿里开发手册1.5.0里面一丶(六)也讲到,很多关于线程的规范,下面列举几条:
- 线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。
-
线程池不允许使用 Executors 去创建，而是通过ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
说明：Executors 返回的线程池对象的弊端如下：
1） FixedThreadPool 和 SingleThreadPool：
允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求,从而导致 OOM。
2） CachedThreadPool：
允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程,从而导致 OOM。
- 创建线程或线程池时请指定有意义的线程名称，方便出错时回溯。
测试小例子:
public class T05_00_HelloThreadPool {
static class Task implements Runnable {
private int i;
public Task(int i) {
this.i = i;
}
@Override
public void run() {
// 打印一下当前线程
System.out.println(Thread.currentThread().getName() + " Task " + i);
try {
// 阻塞住,以便认识不同的拒绝策略
TimeUnit.DAYS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return "Task{" +
"i=" + i +
'}';
}
}
public static void main(String[] args) {
// 初始化一个线程池,最多同时接纳8个任务
ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4,
60, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(4),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy()
);
// 把线程池占满
for (int i = 0; i < 8; i++) {
tpe.execute(new Task(i));
}
// 打印一下当前线程等待队列,是线程2,3,4,5;
// 因为0和1正被核心线程执行,6和7被非核心线程执行
System.out.println(tpe.getQueue());
tpe.execute(new Task(100));
// 如果是DiscardOldestPolicy,会发现任务2被丢掉了,任务100加入等待队列
// 如果是CallerRunsPolicy,这句话不会打印,因为新的任务被主线程执行,而任务会阻塞线程;但是会打印main Task 100
System.out.println("main thread end\n" + tpe.getQueue());
// 不再接收新任务,等已有任务执行完后关掉线程池
tpe.shutdown();
// 尝试马上关掉线程池,不等当前任务结束,而是通过Thread.interrupt打断线程
// 如果线程没有正确处理InterruptedException,那就永远那不会被终结
// tpe.shutdownNow();
}
}
调整线程池的大小
下面是一个建议,也可以说是标准公式吧,但是这个公式中的等待时间和预估时间的比率很难预估出来,工程中还是需要经过各种情况的压力测试,然后取一个相对各方面都照顾的到的值.
一般的等待时间都花在IO上,所以W/C比较高时也称为IO密集型.
ThreadPoolExecutor源码解析
这块扣起来贼头疼,我们先领会战略精神,具体战术日后再议…
昨天看AQS的源码,扣了半天没搞明白,浪费好多时间,还有很多"上天入地"的任务待完成…
这里补充记录一点,JAVA(不知道其他语言怎么说…)中整数的表示形式,以4位的数来说:
1.正整数和0,就是正常的二进制,1就是0001,2就是0010
2.负整数=对应正整数的反码的补码,-1反码->1110补码->1111,即十进制-1的二进制为1111
1、常用变量的解释
// 1. `ctl`，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 2. `COUNT_BITS`，`Integer.SIZE`为32，所以`COUNT_BITS`为29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 3. `CAPACITY`，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
// 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0
private static int workerCountOf(int c) { return c & CAPACITY; }
// 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
// 8. `runStateLessThan()`，线程池状态小于xx
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 9. `runStateAtLeast()`，线程池状态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
2、构造方法
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
// 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
3、提交执行task的过程
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();
// worker数量比核心线程数小，直接创建worker执行任务
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// worker数量超过核心线程数，任务直接进入队列
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
// 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
if (! isRunning(recheck) && remove(command))
reject(command);
// 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
// 这儿有3点需要注意：
// 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
// 2. addWorker第2个参数表示是否创建核心线程
// 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
else if (!addWorker(command, false))
reject(command);
}
4、addworker源码解析
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 外层自旋
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
// (rs > SHUTDOWN) ||
// (rs == SHUTDOWN && firstTask != null) ||
// (rs == SHUTDOWN && workQueue.isEmpty())
// 1. 线程池状态大于SHUTDOWN时，直接返回false
// 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
// 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内层自旋
for (;;) {
int wc = workerCountOf(c);
// worker数量超过容量，直接返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 使用CAS的方式增加worker数量。
// 若增加成功，则直接跳出外层循环进入到第二部分
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
// 线程池状态发生变化，对外层循环进行自旋
if (runStateOf(c) != rs)
continue retry;
// 其他情况，直接内层循环进行自旋即可
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
// worker的添加必须是串行的，因此需要加锁
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 这儿需要重新检查线程池状态
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// worker已经调用过了start()方法，则不再创建worker
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
// worker创建并添加到workers成功
workers.add(w);
// 更新`largestPoolSize`变量
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 启动worker线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
5、线程池worker任务单元
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// 省略代码...
}
6、核心线程执行逻辑-runworker
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
// 调用unlock()是为了让外部可以中断
w.unlock(); // allow interrupts
// 这个变量用于判断是否进入过自旋（while循环）
boolean completedAbruptly = true;
try {
// 这儿是自旋
// 1. 如果firstTask不为null，则执行firstTask；
// 2. 如果firstTask为null，则调用getTask()从队列获取任务。
// 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
while (task != null || (task = getTask()) != null) {
// 这儿对worker进行加锁，是为了达到下面的目的
// 1. 降低锁范围，提升性能
// 2. 保证每个worker执行的任务是串行的
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
// 如果线程池正在停止，则对当前线程进行中断操作
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
// 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
// 这两个方法在当前类里面为空实现。
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
// 帮助gc
task = null;
// 已完成任务数加一
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 自旋操作被退出，说明线程池正在结束
processWorkerExit(w, completedAbruptly);
}
}
![Java小案例——随机数猜测随机数猜测[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)