项目中经常有些任务需要异步(提交到线程池中)去执行，而主线程往往需要知道异步执行产生的结果，这时我们要怎么做呢？用runnable是无法实现的，我们需要用callable看下面的代码：
import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; public class AddTask implements Callable{ private int a,b; public AddTask(int a, int b) { this.a = a; this.b = b; } @Override public Integer call throws Exception { Integer result = a + b; return result; } public static void main(String[] args) throws InterruptedException, ExecutionException { ExecutorService executor = Executors.newSingleThreadExecutor; //JDK目前为止返回的都是FutureTask的实例 Future future = executor.submit(new AddTask(1, 2)); Integer result = future.get;// 只有当future的状态是已完成时(future.isDOne= true),get方法才会返回 } }
虽然可以实现获取异步执行结果的需求，但是我们发现这个Future其实很不好用，因为它没有提供通知的机制，也就是说我们不知道future什么时候完成（如果我们需要轮询isDone()来判断的话感觉就没有用这个的必要了）。看下java.util.concurrent.future.Future 的接口方法：
public interface Future{ boolean cancel(boolean mayInterruptIfRunning); boolean isCancelled; boolean isDone; V get throws InterruptedException, ExecutionException; V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
由此可见JDK的Future机制其实并不好用，如果能给这个future加个监听器，让它在完成时通知监听器的话就比较好用了，就像下面这个IFuture：
package future; import java.util.concurrent.CancellationException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; /** * The result of an asynchronous operation. * * @author lixiaohui * @param执行结果的类型参数 */ public interface IFuture extends Future { boolean isSuccess; // 是否成功 V getNow; //立即返回结果(不管Future是否处于完成状态) Throwable cause; //若执行失败时的原因 boolean isCancellable; //是否可以取消 IFuture await throws InterruptedException; //等待future的完成 boolean await(long timeoutMillis) throws InterruptedException; // 超时等待future的完成 boolean await(long timeout, TimeUnit timeunit) throws InterruptedException; IFuture awaitUninterruptibly; //等待future的完成，不响应中断 boolean awaitUninterruptibly(long timeoutMillis);//超时等待future的完成，不响应中断 boolean awaitUninterruptibly(long timeout, TimeUnit timeunit); IFuture addListener(IFutureListener l); //当future完成时，会通知这些加进来的监听器 IFuture removeListener(IFutureListener l); }
接下来就一起来实现这个IFuture，在这之前要说明下Object.wait,Object.notifyAll方法，因为整个Future实现的原&＃65533;&＃65533;&＃65533;的核心就是这两个方法.看看JDK里面的解释:
public class Object {
/**
* Causes the current thread to wait until another thread invokes the
* {@link java.lang.Object#notify} method or the
* {@link java.lang.Object#notifyAll} method for this object.
* In other words, this method behaves exactly as if it simply
* performs the call {@code wait(0)}.
* 调用该方法后，当前线程会释放对象监视器锁，并让出CPU使用权。直到别的线程调用notify/notifyAll
*/
public final void wait throws InterruptedException {
wait(0);
}
/**
* Wakes up all threads that are waiting on this object's monitor. A
* thread waits on an object's monitor by calling one of the
* {@code wait} methods.
*
* The awakened threads will not be able to proceed until the current
* thread relinquishes the lock on this object. The awakened threads
* will compete in the usual manner with any other threads that might
* be actively competing to synchronize on this object; for example,
* the awakened threads enjoy no reliable privilege or disadvantage in
* being the next thread to lock this object.
*/
public final native void notifyAll;
}
知道这个后，我们要自己实现Future也就有了思路，当线程调用了IFuture.await等一系列的方法时，如果Future还未完成，那么就调用future.wait 方法使线程进入WAITING状态。而当别的线程设置Future为完成状态（注意这里的完成状态包括正常结束和异常结束）时，就需要调用future.notifyAll方法来唤醒之前因为调用过wait方法而处于WAITING状态的那些线程。完整的实现如下(代码应该没有很难理解的地方，我是参考netty的Future机制的。有兴趣的可以去看看netty的源码)：
package future;
import java.util.Collection;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
*
* 正常结束时, 若执行的结果不为null, 则result为执行结果; 若执行结果为null, 则result = {@link AbstractFuture#SUCCESS_SIGNAL}
* 异常结束时, result为 {@link CauseHolder} 的实例;若是被取消而导致的异常结束, 则result为 {@link CancellationException} 的实例, 否则为其它异常的实例
* 以下情况会使异步操作由未完成状态转至已完成状态, 也就是在以下情况发生时调用notifyAll方法:
* 那么要怎么使用这个呢，有了上面的骨架实现，我们就可以定制各种各样的异步结果了。下面模拟一下一个延时的任务：
package future.test;
import future.IFuture;
import future.IFutureListener;
/**
* 延时加法
* @author lixiaohui
*
*/
public class DelayAdder {
public static void main(String[] args) {
new DelayAdder.add(3 * 1000, 1, 2).addListener(new IFutureListener {
@Override
public void operationCompleted(IFuture future) throws Exception {
System.out.println(future.getNow);
}
});
}
/**
* 延迟加
* @param delay 延时时长 milliseconds
* @param a 加数
* @param b 加数
* @return 异步结果
*/
public DelayAdditionFuture add(long delay, int a, int b) {
DelayAdditionFuture future = new DelayAdditionFuture;
new Thread(new DelayAdditionTask(delay, a, b, future)).start;
return future;
}
private class DelayAdditionTask implements Runnable {
private long delay;
private int a, b;
private DelayAdditionFuture future;
public DelayAdditionTask(long delay, int a, int b, DelayAdditionFuture future) {
super;
this.delay = delay;
this.a = a;
this.b = b;
this.future = future;
}
@Override
public void run {
try {
Thread.sleep(delay);
Integer i = a + b;
// TODO 这里设置future为完成状态(正常执行完毕)
future.setSuccess(i);
} catch (InterruptedException e) {
// TODO 这里设置future为完成状态(异常执行完毕)
future.setFailure(e.getCause);
}
}
}
} package future.test;
import future.AbstractFuture;
import future.IFuture;
//只是把两个方法对外暴露
public class DelayAdditionFuture extends AbstractFuture {
@Override
public IFuture setSuccess(Object result) {
return super.setSuccess(result);
}
@Override
public IFuture setFailure(Throwable cause) {
return super.setFailure(cause);
}
}
可以看到客户端不用主动去询问future是否完成，而是future完成时自动回调operationcompleted方法，客户端只需在回调里实现逻辑即可。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有