创建一个线程,却需要调用操作系统内核的API,然后操作系统要为线程分配一系列的资源,这个成本就很高了,所以线程是一个重量级的对象,应该避免频繁创建和销毁。
目前业界线程池的设计,普遍采用的都是生产者-消费者模式。线程池的使用方是生产者,线程池本身是消费者。
在MyThreadPool的内部,我们维护了一个阻塞队列workQueue和一组工作线程,工作线程的个数由构造函数中的poolSize来指定。用户通过调用execute()方法来提交Runnable任务,execute()方法的内部实现仅仅是将任务加入到workQueue中。MyThreadPool内部维护的工作线程会消费workQueue中的任务并执行任务,相关的代码就是代码①处的while循环。
Java并发包里提供的线程池,远比我们上面的示例代码强大得多,当然也复杂得多。Java提供的线程池相关的工具类中,最核心的是ThreadPoolExecutor,通过名字你也能看出来,它强调的是Executor,而不是一般意义上的池化资源。
构造函数
ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
考虑到ThreadPoolExecutor的构造函数实在是有些复杂,所以Java并发包里提供了一个线程池的静态工厂类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()方法没有返回值)
如何获取任务执行结果
ThreadPoolExecutor的void execute(Runnable command) 方法,利用这个方法虽然可以提交任务,但是却没有办法获取任务的执行结果(execute()方法没有返回值)。
// 提交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个方法,我都列在下面了,它们分别是
取消任务的方法cancel()、
判断任务是否已取消的方法isCancelled()、
判断任务是否已结束的方法isDone()
2个获得任务执行结果的get()和get(timeout, unit),
get(timeout, unit)支持超时机制。
通过Future接口的这5个方法你会发现,我们提交的任务不但能够获取任务执行结果,还可以取消任务。不过需要注意的是:这两个get()方法都是阻塞式的,如果被调用的时候,任务还没有执行完,那么调用get()方法的线程会阻塞,直到任务执行完才会被唤醒。
// 取消任务
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<T> task):这个方法的参数是一个Callable接口,它只有一个call()方法,并且这个方法是有返回值的,所以这个方法返回的Future对象可以通过调用其get()方法来获取任务的执行结果。
- 提交Runnable任务及结果引用 submit(Runnable task, T result):这个方法很有意思,假设这个方法返回的Future对象是f,f.get()的返回值就是传给submit()方法的参数result。这个方法该怎么用呢?下面这段示例代码展示了它的经典用法。需要你注意的是Runnable接口的实现类Task声明了一个有参构造函数 Task(Result r) ,创建Task对象的时候传入了result对象,这样就能在类Task的run()方法中对result进行各种操作了。result相当于主线程和子线程之间的桥梁,通过它主子线程可以共享数据。
下面我们再来介绍FutureTask工具类。前面我们提到的Future是一个接口,而FutureTask是一个实实在在的工具类,这个工具类有两个构造函数,它们的参数和前面介绍的submit()方法类似,所以这里我就不再赘述了。
FutureTask(Callable<V> callable);
FutureTask(Runnable runnable, V result);
那如何使用FutureTask呢?其实很简单,FutureTask实现了Runnable和Future接口,由于实现了Runnable接口,所以可以将FutureTask对象作为任务提交给ThreadPoolExecutor去执行,也可以直接被Thread执行;又因为实现了Future接口,所以也能用来获得任务的执行结果。下面的示例代码是将FutureTask对象提交给ThreadPoolExecutor去执行。
// 创建FutureTask
FutureTask<Integer> futureTask = new FutureTask<>(()-> 1+2);
// 创建线程池
ExecutorService es = Executors.newCachedThreadPool();
// 提交FutureTask
es.submit(futureTask);
// 获取计算结果
Integer result = futureTask.get();
FutureTask对象直接被Thread执行的示例代码如下所示。相信你已经发现了,利用FutureTask对象可以很容易获取子线程的执行结果。
// 创建FutureTask
FutureTask<Integer> futureTask
= new FutureTask<>(()-> 1+2);
// 创建并启动线程
Thread T1 = new Thread(futureTask);
T1.start();
// 获取计算结果
Integer result = futureTask.get();
利用Java并发包提供的Future可以很容易获得异步任务的执行结果,无论异步任务是通过线程池ThreadPoolExecutor执行的,还是通过手工创建子线程来执行的。Future可以类比为现实世界里的提货单,比如去蛋糕店订生日蛋糕,蛋糕店都是先给你一张提货单,你拿到提货单之后,没有必要一直在店里等着,可以先去干点其他事,比如看场电影;等看完电影后,基本上蛋糕也做好了,然后你就可以凭提货单领蛋糕了。
利用多线程可以快速将一些串行的任务并行化,从而提高性能;如果任务之间有依赖关系,比如当前任务依赖前一个任务的执行结果,这种问题基本上都可以用Future来解决。在分析这种问题的过程中,建议你用有向图描述一下任务之间的依赖关系,同时将线程的分工也做好,类似于烧水泡茶最优分工方案那幅图。
网友评论