文章目录
- 概述
- 概述
- Executor与线程池
- Java 中的线程池
- 使用线程池的注意事项
- 强烈建议使用有界队列
- 默认拒绝策略要慎重使用
- 注意异常处理的问题
- 如何获取任务执行结果
概述
最近在阅读耗子叔的《左耳听风》 , 记一些小笔记
概述
在 Java 中,在 JDK 1.8 里也引入了类似 JavaScript 的玩法 —— CompletableFuture
。这个类提供了大量的异步编程中 Promise 的各种方式。
CompletableFuture.supplyAsync(this::findReceiver)
.thenApply(this::sendMsg)
.thenAccept(this::notify);
supplyAsync()
表示执行一个异步方法thenApply()
表示执行成功后再串联另外一个异步方法- 最后是
thenAccept()
来处理最终结果。
接下来,我们再来看一下,Java 这个类相关的异常处理:
CompletableFuture.supplyAsync(Integer::parseInt) //输入: "ILLEGAL"
.thenApply(r -> r * 2 * Math.PI)
.thenApply(s -> "apply>> " + s)
.exceptionally(ex -> "Error: " + ex.getMessage());
注意到上面代码里的 exceptionally() 方法 , 运行上面的代码,会出现如下输出:
Error: java.lang.NumberFormatException: For input string: "ILLEGAL"
也可以这样:
CompletableFuture.supplyAsync(Integer::parseInt) // 输入: "ILLEGAL"
.thenApply(r -> r * 2 * Math.PI)
.thenApply(s -> "apply>> " + s)
.handle((result, ex) -> {
if (result != null) {
return result;
} else {
return "Error handling: " + ex.getMessage();
}
});
上面代码中,使用了 handle() 方法来处理最终的结果,其中包含了异步函数中的错误处理。
Executor与线程池
创建对象,仅仅是在 JVM 的堆里分配一块内存而已;而创建一个线程,却需要调用操作系统内核的 API,然后操作系统要为线程分配一系列的资源,这个成本就很高了,所以线程是一个重量级的对象,应该避免频繁创建和销毁
为什么线程池没有采用一般意义上池化资源的设计方法呢?如果线程池采用一般意义上池化资源的设计方法,应该是下面示例代码这样。
//采用一般意义上池化资源的设计方法
class ThreadPool{
// 获取空闲线程
Thread acquire() {
}
// 释放线程
void release(Thread t){
}
}
//期望的使用
ThreadPool pool;
Thread T1=pool.acquire();
//传入Runnable对象
T1.execute(()->{
//具体业务逻辑
......
});
可以来思考一下,假设我们获取到一个空闲线程 T1,然后该如何使用 T1 呢?
我们期望的可能是这样:通过调用 T1 的 execute()
方法,传入一个 Runnable
对象来执行具体业务逻辑,就像通过构造函数 Thread(Runnable target)
创建线程一样。可惜的是,翻遍 Thread 对象的所有方法,都不存在类似 execute(Runnable target)
这样的公共方法。
线程池是一种生产者 - 消费者模式。 线程池的使用方是生产者,线程池本身是消费者。在下面的示例代码中,创建了一个非常简单的线程池 MyThreadPool
,可以通过它来理解线程池的工作原理。
//简化的线程池,仅用来说明工作原理
class MyThreadPool{
//利用阻塞队列实现生产者-消费者模式
BlockingQueue<Runnable> workQueue;
//保存内部工作线程
List<WorkerThread> threads = new ArrayList<>();
// 构造方法
MyThreadPool(int poolSize, BlockingQueue<Runnable> workQueue){
this.workQueue = workQueue;
// 创建工作线程
for(int idx=0; idx<poolSize; idx++){
WorkerThread work = new WorkerThread();
work.start();
threads.add(work);
}
}
// 提交任务
void execute(Runnable command){
workQueue.put(command);
}
// 工作线程负责消费任务,并执行任务
class WorkerThread extends Thread{
public void run() {
//循环取任务并执行
while(true){ // ①
Runnable task = workQueue.take();
task.run();
}
}
}
}
/** 下面是使用示例 **/
// 创建有界阻塞队列
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(2);
// 创建线程池
MyThreadPool pool = new MyThreadPool( 10, workQueue);
// 提交任务
pool.execute(()->{
System.out.println("hello");
});
在 MyThreadPool 的内部,我们维护了一个阻塞队列 workQueue
和一组工作线程,工作线程的个数由构造函数中的 poolSize
来指定。用户通过调用 execute()
方法来提交 Runnable
任务,execute() 方法的内部实现仅仅是将任务加入到 workQueue 中。MyThreadPool 内部维护的工作线程会消费 workQueue 中的任务并执行任务,相关的代码就是代码①处的 while 循环。线程池主要的工作原理就这些.
Java 中的线程池
Java 提供的线程池相关的工具类中,最核心的是 ThreadPoolExecutor
,通过名字你也能看出来,它强调的是 Executor,而不是一般意义上的池化资源。
最完备的构造函数有 7 个参数
ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
可以把线程池类比为一个项目组,而线程就是项目组的成员
corePoolSize
:表示线程池保有的最小线程数。有些项目很闲,但是也不能把人都撤了,至少要留 corePoolSize 个人坚守阵地。maximumPoolSize
:表示线程池创建的最大线程数。当项目很忙时,就需要加人,但是也不能无限制地加,最多就加到 maximumPoolSize 个人。当项目闲下来时,就要撤人了,最多能撤到 corePoolSize 个人。keepAliveTime & unit
:上面提到项目根据忙闲来增减人员,那在编程世界里,如何定义忙和闲呢?很简单,一个线程如果在一段时间内,都没有执行任务,说明很闲,keepAliveTime 和 unit 就是用来定义这个“一段时间”的参数。也就是说,如果一个线程空闲了keepAliveTime & unit这么久,而且线程池的线程数大于 corePoolSize ,那么这个空闲的线程就要被回收了workQueue
:工作队列,和上面示例代码的工作队列同义。threadFactory
:通过这个参数可以自定义如何创建线程,handler
:通过这个参数你可以自定义任务的拒绝策略。如果线程池中所有的线程都在忙碌,并且工作队列也满了(前提是工作队列是有界队列),那么此时提交任务,线程池就会拒绝接收。至于拒绝的策略,可以通过 handler 这个参数来指定。
ThreadPoolExecutor
已经提供了以下 4 种策略。
CallerRunsPolicy:提交任务的线程自己去执行该任务。
AbortPolicy:默认的拒绝策略,会 throws RejectedExecutionException。
DiscardPolicy:直接丢弃任务,没有任何异常抛出。
DiscardOldestPolicy:丢弃最老的任务,其实就是把最早进入工作队列的任务丢弃,然后把新任务加入到工作队列。
Java 在 1.6 版本还增加了 allowCoreThreadTimeOut(boolean value)
方法,它可以让所有线程都支持超时,这意味着如果项目很闲,就会将项目组的成员都撤走。
使用线程池的注意事项
强烈建议使用有界队列
考虑到 ThreadPoolExecutor 的构造函数实在是有些复杂,所以 Java 并发包里提供了一个线程池的静态工厂类 Executors,利用 Executors 你可以快速创建线程池。不过目前大厂的编码规范中基本上都不建议使用 Executors 了。
不建议使用 Executors 的最重要的原因是:Executors 提供的很多方法默认使用的都是无界的 LinkedBlockingQueue,高负载情境下,无界队列很容易导致 OOM,而 OOM 会导致所有请求都无法处理,这是致命问题。所以强烈建议使用有界队列。
默认拒绝策略要慎重使用
使用有界队列,当任务过多时,线程池会触发执行拒绝策略,线程池默认的拒绝策略会 throw RejectedExecutionException 这是个运行时异常,对于运行时异常编译器并不强制 catch 它,所以开发人员很容易忽略。因此默认拒绝策略要慎重使用。如果线程池处理的任务非常重要,建议自定义自己的拒绝策略;并且在实际工作中,自定义的拒绝策略往往和降级策略配合使用。
注意异常处理的问题
使用线程池,还要注意异常处理的问题,例如通过 ThreadPoolExecutor
对象的 execute()
方法提交任务时,如果任务在执行的过程中出现运行时异常,会导致执行任务的线程终止;不过,最致命的是任务虽然异常了,但是你却获取不到任何通知,这会让你误以为任务都执行得很正常。虽然线程池提供了很多用于异常处理的方法,但是最稳妥和简单的方案还是捕获所有异常并按需处理,你可以参考下面的示例代码。
try {
//业务逻辑
} catch (RuntimeException x) {
//按需处理
} catch (Throwable x) {
//按需处理
}
如何获取任务执行结果
ThreadPoolExecutor
的 void execute(Runnable command)
方法,利用这个方法虽然可以提交任务,但是却没有办法获取任务的执行结果(execute()
方法没有返回值)。而很多场景下,我们又都是需要获取任务的执行结果的。
Java 通过 ThreadPoolExecutor
提供的 3 个 submit()
方法和 1 个 FutureTask
工具类来支持获得任务执行结果的需求。
// 提交Runnable任务
Future<?> submit(Runnable task);
// 提交Callable任务
<T> Future<T> submit(Callable<T> task);
// 提交Runnable任务及结果引用
<T> Future<T> submit(Runnable task, T result);
它们的返回值都是 Future 接口.
Future 接口有 5 个方法
// 取消任务
boolean cancel(boolean mayInterruptIfRunning);
// 判断任务是否已取消
boolean isCancelled();
// 判断任务是否已结束
boolean isDone();
// 获得任务执行结果
get();
// 获得任务执行结果,支持超时
get(long timeout, TimeUnit unit);
这 3 个 submit() 方法之间的区别在于方法参数不同,
- 提交
Runnable
任务submit(Runnable task)
:这个方法的参数是一个 Runnable 接口,Runnable 接口的 run() 方法是没有返回值的,所以 submit(Runnable task) 这个方法返回的 Future 仅可以用来断言任务已经结束了,类似于 Thread.join()。 - 提交
Callable
任务submit(Callable task)
:这个方法的参数是一个 Callable 接口,它只有一个 call() 方法,并且这个方法是有返回值的,所以这个方法返回的 Future 对象可以通过调用其 get() 方法来获取任务的执行结果。