线程池快速入门及线程复用原理
站长
· 阅读数 79
线程池快速入门及线程复用原理
线程池-ThreadPoolExecutor
常用方法
- execute():提交任务
- submit():提交任务，能够返回执行结果，获取异常。
- shutdown():关闭线程池，等待任务都执行完。
- shutdownNow():关闭线程池，不等待任务执行完。
- getTaskCount():线程池执行任务总数（已执行+未执行）
- getCompletedTaskCount(): 已完成的任务数量。
- getPoolSize(): 线程池当前的线程数量。
- getActiveCount():当前线程池中正在执行任务的线程数量。
submit()提交方法demo
ExecutorService executorService1 = Executors.newFixedThreadPool(1);
Runnable runnable =() -> {
System.out.println("run");
};
Future future = executorService1.submit(runnable);
try {
future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
Callable callable = () -> {
return 1;
};
Future future1 = executorService1.submit(callable);
try {
Integer a = (Integer) future1.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
Future 接口有 5 个方法
- cancel()：取消任务的方法
- isCancelled()：判断任务是否已取消的方法
- isDone()：判断任务是否已结束的方法
- get() ：获得任务执行结果
- get(timeout, unit)：获得任务执行结果
从网上看到的截图
线程池执行任务规则
默认情况下，新创建的线程池中没有任何线程，等任务来才会创建线程去执行任务。如果想一创建线程池就创建线程可是使用prestartCoreThread()和prestartAllCoreThreads()。一般在需要预热的系统中可能用到。
-
prestartCoreThread()：启动一个核心线程，使其闲置地等待工作。 这将覆盖仅在执行新任务时启动核心线程的默认策略。 如果所有核心线程已经启动，则此方法将返回false 。
-
prestartAllCoreThreads()：启动所有核心线程，使它们无所事事地等待工作。 这将覆盖仅在执行新任务时启动核心线程的默认策略。返回启动的线程数
线程等待keepAliveTime后仍然没有给他分配新的任务，线程就会被回收。不分核心线程和非核心线程，直到当前线程数等于corePoolSize才会停止回收。最后会剩余corePoolSize 数的线程一直存在，如果想把这些空闲的线程回收需要调用allowCoreThreadTimeOut(boolean value)
/**设置策略，以控制在保持活动时间内没有任务到达时核心线程是否可能超时并终止，并在新任务到达时根据需要替换。 如果为false，则由于缺少传入任务，核心线程永远不会终止。 如果为true，则适用于非核心线程的相同的保持活动策略也适用于核心线程。 为了避免连续更换线程，设置true时，保持活动时间必须大于零。 通常应在主动使用池之前调用此方法。
*/
IllegalArgumentException如果value为true并且当前的保持活动时间不大于零
public void allowCoreThreadTimeOut(boolean value) {
if (value && keepAliveTime <= 0)
throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
if (value != allowCoreThreadTimeOut) {
allowCoreThreadTimeOut = value;
if (value)
interruptIdleWorkers();
}
}
手动创建一个线程池
int corePoolSize = 5;
// 获取cpu核心线程数也就是计算资源。
int availableProcessors = Runtime.getRuntime().availableProcessors();
ExecutorService executorService = new ThreadPoolExecutor(corePoolSize,
availableProcessors,
60L,
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(1000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("myThreadName");
return t;
}
},
new java.util.concurrent.RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
log.error("async sender is error rejected, runnable: {}, executor: {}", r, executor);
}
}
);
关于线程池中最佳线程数量网上有好多资料，但是最后都要求根据系统自己调试，例如
- cpu密集型（加密、hash计算）：最佳线程数为cpu核心数的1-2倍。
- 耗时IO型（读写数据库、文件、网络请求等）：最佳线程数一般大于cpu核心数多倍，以JVM线程监控显示繁忙情况为依据。参考Brain Goetz推荐的计算方法。线程数=CPU核心数*（1+平均等待时间、平均工作时间） cpu核心数可以使用
Runtime.getRuntime().availableProcessors()获取。
本人没有真的研究，记得唯一的一次调优还是实习那会做联通的某app，甲方验收要看性能，跟着老大搞了几个晚上。后来的项目没有用到过线程池。
手动创建线程池的例子上边的demo已经写了。下面说说使用Executors工具类创建线程池。
阿里巴巴开发手册并发编程中不允许使用Executors创建线程池，而是通过ThreadPoolExecutor的方式。
Executor可以创建四种线程池
- newCachedThreadPool：创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。缓冲队列是
SynchronousQueue，不存储元素，核心线程不够用了就创建非核心线程。 - newFixedThreadPool：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。缓冲队列LinkedBlockingQueue是长度为
2147483647的队列，几乎相当于无界。可能会引发OOM。 - newScheduledThreadPool： 创建一个定长线程池，支持定时及周期性任务执行。
- newSingleThreadExecutor： 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。缓冲队列LinkedBlockingQueue是长度为
2147483647的队列，几乎相当于无界。可能会引发OOM。
停止线程池相关方法
-
shutdown：将线程池状态设置为shutdown。拒绝新提交的任务，会将正在执行的任务和等待队列中的任务执行完成，没有返回值
-
isShutdown：返回这个线程池的状态。
-
isTerminated：是否终止
-
awaitTermination：等待一段事件，最后关闭返回true、超时返回false。
/** * Blocks until all tasks have completed execution after a shutdown * request, or the timeout occurs, or the current thread is * interrupted, whichever happens first. * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument * @return {@code true} if this executor terminated and * {@code false} if the timeout elapsed before termination * @throws InterruptedException if interrupted while waiting */ boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; -
shutdownNow：拒绝新提交的任务，对正在执行的任务尝试interrupt中断，等待队列中的任务不执行并返回未执行列表。
停止线程池中的任务
// todo
异常处理
使用线程池，还要注意异常处理的问题，例如通过 ThreadPoolExecutor 对象的 execute() 方法提交任务时，如果任务在执行的过程中出现运行时异常，会导致执行任务的线程终止；不过，最致命的是任务虽然异常了，但是你却获取不到任何通知，这会让你误以为任务都执行得很正常。虽然线程池提供了很多用于异常处理的方法，但是最稳妥和简单的方案还是捕获所有异常并按需处理。例如
try {
// 业务逻辑
} catch (RuntimeException x) {
// 按需处理
} catch (Throwable x) {
// 按需处理
}
线程池中线程怎么实现复用的
先大概描述一下，线程池把线程和任务解耦，摆脱了之前通过Thread创建线程时，一个线程对应一个任务的限制。线程池中的线程从缓冲队列中拿取任务来执行。如何做到的。线程池创建线程后，让每个线程去执行一个循环任务，while循环检查缓冲队列种中是否有任务去执行，有则调用这个任务的run方法。描述的可能不太清楚，下面贴代码。
先看ThreadPoolExecutor.execute()
public void execute(Runnable command) {
if (command == null) {
throw new NullPointerException();
} else {
int c = this.ctl.get();
if (workerCountOf(c) < this.corePoolSize) {
if (this.addWorker(command, true)) {
return;
}
c = this.ctl.get();
}
// 执行到这里说明当前线程数>=核心线程数或者addWorker失败了。如果线程池状态是 Running 就把任务放入任务队列中。
if (isRunning(c) && this.workQueue.offer(command)) {
int recheck = this.ctl.get();
if (!isRunning(recheck) && this.remove(command)) {
this.reject(command);
} else if (workerCountOf(recheck) == 0) { //执行到这里要防止没有可执行的线程发生
this.addWorker((Runnable)null, false);
}
} else if (!this.addWorker(command, false)) {
this.reject(command);
}
}
}
worker数 < 核心线程数就创建worker，否则就添加到阻塞队列this.workQueue.offer(command)。
下面看addWorker方法
根据上边的execute方法中调用addWorker的方法，可以推测出，当前线程小于核心线程数为true，大于则为false。addWorker()返回true代表添加成功，false代表添加失败。
最终调用t.start()
大概浏览一下代码。
private boolean addWorker(Runnable firstTask, boolean core) {
int c = this.ctl.get();
label247:
while(!runStateAtLeast(c, 0) || !runStateAtLeast(c, 536870912) && firstTask == null && !this.workQueue.isEmpty()) {
while(workerCountOf(c) < ((core ? this.corePoolSize : this.maximumPoolSize) & 536870911)) {
if (this.compareAndIncrementWorkerCount(c)) {
boolean workerStarted = false;
boolean workerAdded = false;
ThreadPoolExecutor.Worker w = null;
try {
// 我在下边贴出了Worker的初始化详细方法
w = new ThreadPoolExecutor.Worker(firstTask);
Thread t = w.thread;
if (t != null) {
ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int c = this.ctl.get();
if (isRunning(c) || runStateLessThan(c, 536870912) && firstTask == null) {
if (t.getState() != State.NEW) {
throw new IllegalThreadStateException();
}
this.workers.add(w);
workerAdded = true;
int s = this.workers.size();
if (s > this.largestPoolSize) {
this.largestPoolSize = s;
}
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
// 调用线程的start方法。
t.start();
workerStarted = true;
}
}
} finally {
if (!workerStarted) {
this.addWorkerFailed(w);
}
}
return workerStarted;
}
c = this.ctl.get();
if (runStateAtLeast(c, 0)) {
continue label247;
}
}
return false;
}
return false;
}
Worker初始化方法
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
this.setState(-1);
// firstTask类型是Runnable
this.firstTask = firstTask;
// 把 Worker 作为 thread 运行的任务
this.thread = ThreadPoolExecutor.this.getThreadFactory().newThread(this);
}
}
再来看Worker.run()
public void run() {
ThreadPoolExecutor.this.runWorker(this);
}
final void runWorker(ThreadPoolExecutor.Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
while(task != null || (task = this.getTask()) != null) {
w.lock();
if ((runStateAtLeast(this.ctl.get(), 536870912) || Thread.interrupted() && runStateAtLeast(this.ctl.get(), 536870912)) && !wt.isInterrupted()) {
wt.interrupt();
}
try {
this.beforeExecute(wt, task);
try {
task.run();
this.afterExecute(task, (Throwable)null);
} catch (Throwable var14) {
this.afterExecute(task, var14);
throw var14;
}
} finally {
task = null;
++w.completedTasks;
w.unlock();
}
}
completedAbruptly = false;
} finally {
this.processWorkerExit(w, completedAbruptly);
}
}
while循环调用Runnable类型的task的run()，而不是创建新的线程 。task不为空就循环，并且不断的从getTask()获取新的任务，继续看getTask()方法。
getTask()方法
private Runnable getTask() {
boolean timedOut = false;
while(true) {
int c = this.ctl.get();
if (runStateAtLeast(c, 0) && (runStateAtLeast(c, 536870912) || this.workQueue.isEmpty())) {
this.decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
boolean timed = this.allowCoreThreadTimeOut || wc > this.corePoolSize;
if (wc <= this.maximumPoolSize && (!timed || !timedOut) || wc <= 1 && !this.workQueue.isEmpty()) {
try {
Runnable r = timed ? (Runnable)this.workQueue.poll(this.keepAliveTime, TimeUnit.NANOSECONDS) : (Runnable)this.workQueue.take();
if (r != null) {
return r;
}
timedOut = true;
} catch (InterruptedException var6) {
timedOut = false;
}
} else if (this.compareAndDecrementWorkerCount(c)) {
return null;
}
}
}
从阻塞队列中拿任务。Runnable r = timed ? (Runnable)this.workQueue.poll(this.keepAliveTime, TimeUnit.NANOSECONDS) : (Runnable)this.workQueue.take();
看到这里就说完了，总结一下就是：新建一个Worker内部类就会建一个线程，并且会把这个内部类本身传进去当作任务去执行，这个内部类的run方法里实现了一个while循环，当任务队列没有任务时结束这个循环，则这个线程就结束。
转载自:https://juejin.cn/post/7238132993533280316