第九章Java异步编程

116 阅读 0 评论 77 点赞

我是靠谱客的博主贤惠信封，这篇文章主要介绍第九章Java异步编程，现在分享给大家，希望可以做个参考。

方法	适用条件及注意事项	原理	备注
public static ExecutorService newCachedThreadPool()	适合用于执行大量耗时较短且提交比较频繁的任务。如果提交的任务执行耗时较长，那么可能导致线程池中的工作者线程无限制地增长，最后导致过多的上下文切换，从而使得整个系统变慢	该方法的返回值相当于：new ThreadPoolExecutor(0,Integer.MAX_VALUE,60L,TimeUnit.SECONDS,new SynchronousQueue<Runnable>());即核心线程池大小为0，最大线程池大小不受限，工作者线程允许的最大空闲时间(keepAliveTime)为60秒，内部以SynchronousQueue为工作者队列的一个线程池	由于是SynchronousQueue，因此必须有take(),才能执行offer()，否则会入队失败，因此在该线程池中的所有工作者线程都在执行任务，即无空闲工作者线程的情况下给其提交任务会导致该任务无法被缓存成功。而ThreadPoolExecutor在任务缓存失败且线程池当前大小未达到最大线程池大小的情况下会创建并启动新的工作者线程。因此，如果任务执行慢，就会有越来越多的线程被创建。
public static ExecutorService newCachedThreadPool(ThreadFactory threadfactory)
public static ExecutorService newFixedThreadPool(int nThreads)	由于该方法返回的线程池的核心线程池大小等于其最大线程池大小，因此该线程池中的工作者线程永远不会超时。我们必须在不再需要该线程池时主动将其关闭	该方法的返回值相当于:new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());这是一种线程池大小一旦达到其核心线程池大小就既不会增加也不会减少工作者线程的固定大小的线程池。
public static ExecutorService newFixedThreadPool(int nThreads,ThreadFactory threadFactory)
public static ExecutorService newSingleThreadExecutor()	适合用于实现单（多）生产者——单消费者模式。该方法的返回值无法被转换为ThreadPoolExecutor类型	该方法的返回值基本相当于Executors.newFixedThreadPool(1)所返回的线程池	该线程池确保了在任意一个时刻只有一个任务会被执行，这就形成了类似锁将原本并发的操作改为串行的操作的结果。
public static ExecutorService newSingleThreadExecutor(ThreadFacotry threadFactory)	适合用于实现单（多）生产者——单消费者模式。该方法的返回值无法被转换为ThreadPoolExecutor类型	该方法的返回值基本相当于Executors.newFixedThreadPool(1)所返回的线程池	该线程池确保了在任意一个时刻只有一个任务会被执行，这就形成了类似锁将原本并发的操作改为串行的操作的结果。

ExecutorCompletionService每执行完一个异步任务，就将该任务对应的Future实例存入其内部维护的BlockingQueue实例之中，而其客户端代码则可以通过ExecutorCompletionService.take()调用来获取这个Future实例。
ExecutorService.invokeAll(Collection<? extends Callable<T>> tasks)也能够用来批量提交异步任务，该方法能够并发执行tasks参数所指定的一批任务，但是该方法只有在tasks参数所指定的一批任务中的所有任务都执行结束之后才返回，其返回值是一个包含各个任务对应的Future实例的列表（List）。
FutureTask是Runnable接口的一个实现类，因此FutureTask表示的异步任务可以提交给专门的工作者线程执行，也可以交给Executor实例（比如线程池）执行；FutureTask还能够直接返回其代表的异步任务的处理结果。FutureTask是java.util.concurrent.RunnableFuture接口的一个实现类。由于RunnableFuture接口继承了Future接口和Runnable接口，因此FutureTask既是Runnable接口的实现类也是Future接口的实现类。FutureTask还支持以回调（Callback）的方式处理任务的执行结果。当FutureTask实例所代表的任务执行结束后，FutureTask.done()会被执行。
基于FutureTask的XML异步解析器

复制代码

public class XMLDocumentParser {
public static ParsingTask newTask(InputStream in){
return new ParsingTask(in);
}
public static ParsingTask newTask(URL url) throws IOException{
return newTask(url, 30000, 30000);
}
// 封装对XML解析结果进行处理的回调结果
public abstract static class ResultHandler{
abstract void onSuccess(Document document);
void onError(Throwable e){
e.printStackTrace();
}
}
public static class ParsingTask {
private final InputStream in;
private volatile Executor executor;
private volatile ResultHandler resultHandler;
public ParsingTask(InputStream in, Executor executor,ResultHandler resultHandler){
this.in = in;
this.executor = executor;
this.resultHandler = resultHandler;
}
public ParsingTask(InputStream in){
this(in, null, null);
}
public Future<Document> execute() throws Exception{
FutureTask<Document> ft;
final Callable<Document> task = new Callable<Document>() {
@Override
public Document call() throws Exception {
return doParse(in);
}
};
final Executor theExecutor = executor;
// 解析模式：异步/同步
// true表示theExecutor不为null，那么就是异步
final boolean isAsyncParsing = (null != theExecutor);
final ResultHandler rh;
if (isAsyncParsing && null != (rh = resultHandler)){
// 异步
// 将Callable转成Runnable
ft = new FutureTask<Document>(task){
@Override
// task执行完成之后会执行done
protected void done() {
// 回调ResultHandler的相关方法对XML解析结果进行处理
callbackResultHandler(this, rh);
}
};
} else {
ft = new FutureTask<Document>(task);
}
if (isAsyncParsing){
// 交给Executor执行，以支持异步
theExecutor.execute(ft);
} else {
// 直接（同步）执行
ft.run();
}
return ft;
}
void callbackResultHandler(FutureTask<Document> ft,ResultHandler rh){
// 获取任务结果前判断任务是否被取消
if (ft.isCancelled()){
System.out.println("parsing cancelled.");
return;
}
try {
Document doc = ft.get();
rh.onSuccess(doc);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
rh.onError(e.getCause());
}
}
// 具体处理过程
static Document doParse(InputStream in) throws Exception{
return null;
}
public ParsingTask setExecutor(Executor executor){
this.executor = executor;
return this;
}
public ParsingTask setResultHandler(ResultHandler resultHandler){
this.resultHandler = resultHandler;
return this;
}
}
public static void main(String[] args) {
// 异步
XMLDocumentParser.newTask(url).setExecutor(executor).setResultHandler(
new ResultHandler() {
@Override
void onSuccess(Document document) {
process(document);
}
}
).execute();
// 同步
Future<Document> future;
future = XMLDocumentParser.newTask(url).execute();
process(future.get());
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
public class XMLDocumentParser {
public static ParsingTask newTask(InputStream in){
return new ParsingTask(in);
}
public static ParsingTask newTask(URL url) throws IOException{
return newTask(url, 30000, 30000);
}
// 封装对XML解析结果进行处理的回调结果
public abstract static class ResultHandler{
abstract void onSuccess(Document document);
void onError(Throwable e){
e.printStackTrace();
}
}
public static class ParsingTask {
private final InputStream in;
private volatile Executor executor;
private volatile ResultHandler resultHandler;
public ParsingTask(InputStream in, Executor executor,ResultHandler resultHandler){
this.in = in;
this.executor = executor;
this.resultHandler = resultHandler;
}
public ParsingTask(InputStream in){
this(in, null, null);
}
public Future<Document> execute() throws Exception{
FutureTask<Document> ft;
final Callable<Document> task = new Callable<Document>() {
@Override
public Document call() throws Exception {
return doParse(in);
}
};
final Executor theExecutor = executor;
// 解析模式：异步/同步
// true表示theExecutor不为null，那么就是异步
final boolean isAsyncParsing = (null != theExecutor);
final ResultHandler rh;
if (isAsyncParsing && null != (rh = resultHandler)){
// 异步
// 将Callable转成Runnable
ft = new FutureTask<Document>(task){
@Override
// task执行完成之后会执行done
protected void done() {
// 回调ResultHandler的相关方法对XML解析结果进行处理
callbackResultHandler(this, rh);
}
};
} else {
ft = new FutureTask<Document>(task);
}
if (isAsyncParsing){
// 交给Executor执行，以支持异步
theExecutor.execute(ft);
} else {
// 直接（同步）执行
ft.run();
}
return ft;
}
void callbackResultHandler(FutureTask<Document> ft,ResultHandler rh){
// 获取任务结果前判断任务是否被取消
if (ft.isCancelled()){
System.out.println("parsing cancelled.");
return;
}
try {
Document doc = ft.get();
rh.onSuccess(doc);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
rh.onError(e.getCause());
}
}
// 具体处理过程
static Document doParse(InputStream in) throws Exception{
return null;
}
public ParsingTask setExecutor(Executor executor){
this.executor = executor;
return this;
}
public ParsingTask setResultHandler(ResultHandler resultHandler){
this.resultHandler = resultHandler;
return this;
}
}
public static void main(String[] args) {
// 异步
XMLDocumentParser.newTask(url).setExecutor(executor).setResultHandler(
new ResultHandler() {
@Override
void onSuccess(Document document) {
process(document);
}
}
).execute();
// 同步
Future<Document> future;
future = XMLDocumentParser.newTask(url).execute();
process(future.get());
}
}

AsyncTask抽象类同时实现了Runnable接口和Callable接口。当任务执行结束后，相应的AsyncTask实例的onResult方法会被调用以处理任务的执行结果；当任务执行过程中抛出异常时，相应的AsyncTask实例的onError方法会被调用以处理这个异常。由于AsyncTask在回调onResult、onError的方法的时候不是直接调用而是通过向Executor实例executor提交一个任务进行的，因此AsyncTask的任务执行（即AsyncTask.run()调用）可以是在一个工作者线程中进行的，而对任务执行结果的处理则可以在另外一个线程中进行。
支持重复执行的异步任务抽象类

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
public abstract class AsyncTask<V> implements Runnable, Callable<V> {
protected final Executor executor;
public AsyncTask(Executor executor){
this.executor = executor;
}
public AsyncTask(){
this(new Executor() {
@Override
public void execute(Runnable command) {
command.run();
}
});
}
// run()成为入口
@Override
public void run() {
Exception exp = null;
V r = null;
try{
// 执行task
r = call();
} catch (Exception e){
exp = e;
}
final V result = r;
// 异步处理结果
if (null == exp){
executor.execute(new Runnable() {
@Override
public void run() {
onResult(result);
}
});
} else {
final Exception exceptionCaught = exp;
executor.execute(new Runnable() {
@Override
public void run() {
onError(exceptionCaught);
}
});
}
}
protected abstract void onResult(V result);
protected void onError(Exception e){
e.printStackTrace();;
}
}

周期性任务的执行结果处理Demo

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class PeriodicTaskResultHandlingDemo {
final static ScheduledExecutorService ses = Executors.newScheduledThreadPool(2);
public static void main(String[] args) throws InterruptedException {
final String host = args[0];
final AsyncTask<Integer> asyncTask = new AsyncTask<Integer>(ses) {
final Random rnd = new Random();
final String targetHost = host;
@Override
public Integer call() throws Exception {
return doCall();
}
private Integer doCall() throws Exception{
Thread.sleep(2000);
Integer r = Integer.valueOf(rnd.nextInt(4));
return r;
}
@Override
protected void onResult(Integer result) {
doSomething(result);
}
private void doSomething(Integer r){
System.out.println(r);
}
};
ses.scheduleAtFixedRate(asyncTask,0,3, TimeUnit.SECONDS);
}
}

小结

从单个任务的角度来看，任务的执行方式是可以同步的，也可以是异步的。同步方式的优点是代码简单、直接，缺点是它往往意味着阻塞，因此不利于系统的吞吐率。异步方式的优点则是它往往意味着非阻塞，因此有利于系统的吞吐率，其代价是相对复杂的代码和额外的开销。阻塞/非阻塞是任务执行方式的属性，它们与任务执行方式没有必然的联系：同步任务既可能是阻塞的，也可能是非阻塞的；异步任务既可能是非阻塞的，也可能是阻塞的。对于同一个任务，我们既可以说它是同步任务也可以说它是异步任务，这取决于任务的执行方式以及我们的观察角度。
Runnable/Callable接口是对任务处理逻辑进行的抽象，而Executor接口是对任务的执行进行的抽象。Executor接口使得我们能够对任务的提交与任务的具体执行细节进行解耦，这为更改任务的具体执行细节提供了灵活性与便利。ExecutorService接口是对Executore接口的增强：它支持返回异步任务的处理结果、支持资源的管理接口、支持批量任务提交等。ThreadPoolExecutor是Executor/ExecutorService接口的一个实现类。实用工具类Executors为线程池的创建提供了快捷方法。CompletionService接口为异步任务的批量提交以及获取这些任务的处理结果提供了便利，其默认实现类为ExecutorCompletionService。
FutureTask是Java标准库提供的Future接口实现类，它还实现了Runnable接口。因此，FutureTask可直接用来获取异步任务的处理结果，它还可以交给专门的工作者线程执行，也可以交给Executor实例执行，甚至由当前线程直接执行（同步）。一般来说，FutureTask是一次性使用的，一个FutureTask实例代表的任务只能够被执行一次。如果需要多次执行同一个任务，那么可以考虑本书介绍的AsyncTask类.
计划任务的执行方式包括延迟执行和周期性执行。ScheduledThreadPoolExecutor是ScheduledExecutorService接口的默认实现类，它可以用于执行计划任务。ScheduledFuture接口可用来获取延迟执行的任务的处理结果。如果要获取周期性执行的计划任务的处理结果，可以使用自定义的AsyncTask类。周期性执行的计划任务，其执行周期并不是固定的，而是受任务单次执行耗时的影响：提交给scheduleAtFixedRate方法执行的计划任务，其执行周期为max(Execution Time,period)；提交给scheduleWithFixedDelay方法执行的计划任务，其执行周期为Execution Time + delay.计划任务在执行过程中如果抛出未捕获的异常，那么该任务将不会被执行。