第九章Java异步编程

1. ExecutorCompletionService每执行完一个异步任务，就将该任务对应的Future实例存入其内部维护的BlockingQueue实例之中，而其客户端代码则可以通过ExecutorCompletionService.take()调用来获取这个Future实例。
2. ExecutorService.invokeAll(Collection<? extends Callable<T>> tasks)也能够用来批量提交异步任务，该方法能够并发执行tasks参数所指定的一批任务，但是该方法只有在tasks参数所指定的一批任务中的所有任务都执行结束之后才返回，其返回值是一个包含各个任务对应的Future实例的列表（List）。
3. 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());
}
}

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

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

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);
}
}

小结

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

最后

以上就是贤惠信封为你收集整理的第九章Java异步编程的全部内容，希望文章能够帮你解决第九章Java异步编程所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。