Java 线程池原理与使用

2019-07-28 本文已影响6人 Tim在路上

在java 中我们会一般要求创建线程必须使用线程池,因为这样可以避免资源消耗,通过重复利用已经创建的线程来降低线程创建和销毁所造成的消耗, 其次当任务到达时任务可以不用等到线程创建就立即执行,最后可以提高线程的管理性,管控线程,不让线程无限制创建.

线程池的创建

我们可以使用 ThreadPoolExecutor 创建线程池

这里我们不适用Executors 来创建线程池,是因为,Executors 使用默认参数不用自定义参数,

自己经常使用Executors提供的工厂方法创建线程池，所以忽略了线程池内部的实现,同时可以规避规避资源耗尽的风险.

new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
milliseconds,runnableTaskQueue, handler);

corePoolSize(线程池的基本大小): 当提交一个任务到线程池时,线程
池会创建一个线程来执行任务,即使其他空闲的基本线程能够执行新任务
也会创建线程,等到需要执行的任务数大于线程池基本大小时就不再创建。
如果调用了线程池的 prestartAllCoreThreads 方法,线程池会提前创建
并启动所有基本线程。
runnableTaskQueue (任务队列):用于保存等待执行的任务的阻塞队列。
可以选择以下几个阻塞队列。

ArrayBlockingQueue:是一个基于数组结构的有界阻塞队列,此队
列按 FIFO(先进先出)原则对元素进行排序。

LinkedBlockingQueue:一个基于链表结构的阻塞队列,此队列按
FIFO (先进先出) 排序元素,吞吐量通常要高于
ArrayBlockingQueue。静态工厂方法
Executors.newFixedThreadPool()使用了这个队列。

SynchronousQueue:一个不存储元素的阻塞队列。每个插入操作必
须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,
吞吐量通常要高于 LinkedBlockingQueue,静态工厂方法
Executors.newCachedThreadPool 使用了这个队列。

PriorityBlockingQueue:一个具有优先级的无限阻塞队列。

maximumPoolSize(线程池最大大小):线程池允许创建的最大线程数。
如果队列满了,并且已创建的线程数小于最大线程数,则线程池会再创建
新的线程执行任务。值得注意的是如果使用了无界的任务队列这个参数就
没什么效果。
ThreadFactory:用于设置创建线程的工厂,可以通过线程工厂给每个创
建出来的线程设置更有意义的名字。
RejectedExecutionHandler(饱和策略): 当队列和线程池都满了,默认 AbortPolicy:直接抛出异常,
还可设置 CallerRunsPolicy:只用调用者所在线程来运行任务。
o DiscardOldestPolicy:丢弃队列里最近的一个任务,并执行当前
任务。
o DiscardPolicy:不处理,丢弃掉。
o 当然也可以根据应用场景需要来实现 RejectedExecutionHandler
接口自定义策略。如记录日志或持久化不能处理的任务。

向线程池提交任务

我们可以使用 execute 提交的任务,但是 execute 方法没有返回值,所以无法判
断任务是否被线程池执行成功。通过以下代码可知 execute 方法输入的任务是一
个 Runnable 类的实例。

threadsPool.execute(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
}
});

可以使用 submit 方法来提交任务,它会返回一个 future,那么我们可以
通过这个 future 来判断任务是否执行成功,通过 future 的 get 方法来获取返回
值, get 方法会阻塞住直到任务完成,而使用 get(long timeout, TimeUnit unit)
方法则会阻塞一段时间后立即返回,这时有可能任务没有执行完。

Future<Object> future = executor.submit(harReturnValuetask);
try {
Object s = future.get();
} catch (InterruptedException e) {
// 处理中断异常
} catch (ExecutionException e) {
// 处理无法执行任务异常
2} finally {
// 关闭线程池
executor.shutdown();
}

线程池的关闭

调用线程池的 shutdown 或 shutdownNow 方法来关闭线程池,
它们
的原理是遍历线程池中的工作线程,然后逐个调用线程的 interrupt 方法来中断
线程,所以无法响应中断的任务可能永远无法终止。

shutdown 和 shutdownNow 的区别

shutdownNow 首先将线程池的状态设置成 STOP,然后尝试停止所有的正在执行或
暂停任务的线程,并返回等待执行任务的列表,而 shutdown 只是将线程池的状
态设置成 SHUTDOWN 状态,然后中断所有没有正在执行任务的线程。

通常调用 shutdown 来关闭线程池,表示任务执行完关闭,如果任务不一定要执行完,
则可以调用 shutdownNow。

线程池执行过程

提交一个任务到线程池,首先判断基本线程池是否已满,没有满不管线程池中有没有空闲线程都创建一个新线程来执行任务, 判断工作队列是否已满,没满则将新提交的任务存储在工作队列, 满了, 判断整个线程池是否已满, 没满, 则创建一个新的工作线程来
执行任务,满了,则交给饱和策略来处理这个任务.

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//如果线程数小于基本线程数,则创建线程并执行当前任务
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
//如线程数大于等于基本线程数或线程创建失败,则将当前任务放到工作队
列中。
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
//如果线程池不处于运行中或任务无法放入队列,并且当前线程数量小于最
大允许的线程数量,
则创建一个线程执行任务。
else if (!addIfUnderMaximumPoolSize(command))
//抛出 RejectedExecutionException 异常
reject(command); // is shutdown or saturated
}
}

工作线程: 线程池创建线程时,会将线程封装成工作线程 Worker,Worker 在执
行完任务后,还会无限循环获取工作队列里的任务来执行。

public void run() {
try {
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}

合理的配置线程池

任务的性质:CPU 密集型任务,IO 密集型任务和混合型任务。
任务的优先级:高,中和低。
任务的执行时间:长,中和短。
任务的依赖性:是否依赖其他系统资源,如数据库连接。

CPU 密集型任务, 配置尽
可能小的线程,如配置 Ncpu+1 个线程的线程池。
IO 密集型任务则由于线程并不
是一直在执行任务,则配置尽可能多的线程,如 2*Ncpu。
混合型的任务,如果
可以拆分,则将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务,只要这两
个任务执行的时间相差不是太大,那么分解后执行的吞吐率要高于串行执行的吞
吐率,如果这两个任务执行时间相差太大,则没必要进行分解。

优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理。它可以
让优先级高的任务先得到执行,需要注意的是如果一直有优先级高的任务提交到
队列里,那么优先级低的任务可能永远不能执行。
线程提交 SQL 建议使用有界队列,有界队列能增加系统的稳定性和预警能力,可以根据需要设
大一点,比如几千.

线程池的监控

通过继承线程池并重写线程池的 beforeExecute,
afterExecute 和 terminated 方法,我们可以在任务执行前,执行后和线程池关
闭前干一些事情。如监控任务的平均执行时间,最大执行时间和最小执行时间等。