当前位置: 首页 > 编程笔记 >

Java多线程之异步Future机制的原理和实现

蓝逸仙
2023-03-14
本文向大家介绍Java多线程之异步Future机制的原理和实现,包括了Java多线程之异步Future机制的原理和实现的使用技巧和注意事项,需要的朋友参考一下

项目中经常有些任务需要异步(提交到线程池中)去执行,而主线程往往需要知道异步执行产生的结果,这时我们要怎么做呢?用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<Integer> {

 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<Integer> 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<V> {
  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 <V> 执行结果的类型参数
 */
public interface IFuture<V> extends Future<V> { 
 boolean isSuccess; // 是否成功 
 V getNow; //立即返回结果(不管Future是否处于完成状态)
 Throwable cause; //若执行失败时的原因
    boolean isCancellable; //是否可以取消
 IFuture<V> await throws InterruptedException; //等待future的完成
 boolean await(long timeoutMillis) throws InterruptedException; // 超时等待future的完成
 boolean await(long timeout, TimeUnit timeunit) throws InterruptedException;
    IFuture<V> awaitUninterruptibly; //等待future的完成,不响应中断
    boolean awaitUninterruptibly(long timeoutMillis);//超时等待future的完成,不响应中断
 boolean awaitUninterruptibly(long timeout, TimeUnit timeunit);
 IFuture<V> addListener(IFutureListener<V> l); //当future完成时,会通知这些加进来的监听器
 IFuture<V> removeListener(IFutureListener<V> l);
 
} 

接下来就一起来实现这个IFuture,在这之前要说明下Object.wait,Object.notifyAll方法,因为整个Future实现的原���的核心就是这两个方法.看看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.
   * <p>
   * 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 html" target="_blank">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;

/**
 * <pre>
 * 正常结束时, 若执行的结果不为null, 则result为执行结果; 若执行结果为null, 则result = {@link AbstractFuture#SUCCESS_SIGNAL}
 * 异常结束时, result为 {@link CauseHolder} 的实例;若是被取消而导致的异常结束, 则result为 {@link CancellationException} 的实例, 否则为其它异常的实例
 * 以下情况会使异步操作由未完成状态转至已完成状态, 也就是在以下情况发生时调用notifyAll方法:
 * <ul>
 * <li>异步操作被取消时(cancel方法)</li>
 * <li>异步操作正常结束时(setSuccess方法)</li>
 * <li>异步操作异常结束时(setFailure方法)</li>
 * </ul>
 * </pre>
 * 
 * @author lixiaohui
 *
 * @param <V>
 * 异步执行结果的类型
 */
public class AbstractFuture<V> implements IFuture<V> {

 protected volatile Object result; // 需要保证其可见性
    /**
     * 监听器集
     */
 protected Collection<IFutureListener<V>> listeners = new CopyOnWriteArrayList<IFutureListener<V>>;

 /**
 * 当任务正常执行结果为null时, 即客户端调用{@link AbstractFuture#setSuccess(null)}时, 
 * result引用该对象
 */
 private static final SuccessSignal SUCCESS_SIGNAL = new SuccessSignal;

 @Override
 public boolean cancel(boolean mayInterruptIfRunning) {
 if (isDone) { // 已完成了不能取消
  return false;
 }

 synchronized (this) {
  if (isDone) { // double check
  return false;
  }
  result = new CauseHolder(new CancellationException);
  notifyAll; // isDone = true, 通知等待在该对象的wait的线程
 }
 notifyListeners; // 通知监听器该异步操作已完成
 return true;
 }
 
 @Override
 public boolean isCancellable {
 return result == null;
 }
 
 @Override
 public boolean isCancelled {
 return result != null && result instanceof CauseHolder && ((CauseHolder) result).cause instanceof CancellationException;
 }

 @Override
 public boolean isDone {
 return result != null;
 }

 @Override
 public V get throws InterruptedException, ExecutionException {
 await; // 等待执行结果

 Throwable cause = cause;
 if (cause == null) { // 没有发生异常,异步操作正常结束
  return getNow;
 }
 if (cause instanceof CancellationException) { // 异步操作被取消了
  throw (CancellationException) cause;
 }
 throw new ExecutionException(cause); // 其他异常
 }

 @Override
 public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
 if (await(timeout, unit)) {// 超时等待执行结果
  Throwable cause = cause;
  if (cause == null) {// 没有发生异常,异步操作正常结束
  return getNow;
  }
  if (cause instanceof CancellationException) {// 异步操作被取消了
  throw (CancellationException) cause;
  }
  throw new ExecutionException(cause);// 其他异常
 }
 // 时间到了异步操作还没有结束, 抛出超时异常
 throw new TimeoutException;
 }

 @Override
 public boolean isSuccess {
 return result == null ? false : !(result instanceof CauseHolder);
 }

 @SuppressWarnings("unchecked")
 @Override
 public V getNow {
 return (V) (result == SUCCESS_SIGNAL ? null : result);
 }

 @Override
 public Throwable cause {
 if (result != null && result instanceof CauseHolder) {
  return ((CauseHolder) result).cause;
 }
 return null;
 }

 @Override
 public IFuture<V> addListener(IFutureListener<V> listener) {
 if (listener == null) {
  throw new NullPointerException("listener");
 }
 if (isDone) { // 若已完成直接通知该监听器
  notifyListener(listener);
  return this;
 }
 synchronized (this) {
  if (!isDone) {
  listeners.add(listener);
  return this;
  }
 }
 notifyListener(listener);
 return this;
 }

 @Override
 public IFuture<V> removeListener(IFutureListener<V> listener) {
 if (listener == null) {
  throw new NullPointerException("listener");
 }

 if (!isDone) {
  listeners.remove(listener);
 }

 return this;
 }

 @Override
 public IFuture<V> await throws InterruptedException {
 return await0(true);
 }

 
 private IFuture<V> await0(boolean interruptable) throws InterruptedException {
 if (!isDone) { // 若已完成就直接返回了
  // 若允许终端且被中断了则抛出中断异常
  if (interruptable && Thread.interrupted) {
  throw new InterruptedException("thread " + Thread.currentThread.getName + " has been interrupted.");
  }

  boolean interrupted = false;
  synchronized (this) {
  while (!isDone) {
   try {
   wait; // 释放锁进入waiting状态,等待其它线程调用本对象的notify/notifyAll方法
   } catch (InterruptedException e) {
   if (interruptable) {
    throw e;
   } else {
    interrupted = true;
   }
   }
  }
  }
  if (interrupted) {
  // 为什么这里要设中断标志位?因为从wait方法返回后, 中断标志是被clear了的, 
  // 这里重新设置以便让其它代码知道这里被中断了。
  Thread.currentThread.interrupt;
  }
 }
 return this;
 }
 
 @Override
 public boolean await(long timeoutMillis) throws InterruptedException {
 return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), true);
 }
 
 @Override
 public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
 return await0(unit.toNanos(timeout), true);
 }

 private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {
 if (isDone) {
  return true;
 }

 if (timeoutNanos <= 0) {
  return isDone;
 }

 if (interruptable && Thread.interrupted) {
  throw new InterruptedException(toString);
 }

 long startTime = timeoutNanos <= 0 ? 0 : System.nanoTime;
 long waitTime = timeoutNanos;
 boolean interrupted = false;

 try {
  synchronized (this) {
  if (isDone) {
   return true;
  }

  if (waitTime <= 0) {
   return isDone;
  }

  for (;;) {
   try {
   wait(waitTime / 1000000, (int) (waitTime % 1000000));
   } catch (InterruptedException e) {
   if (interruptable) {
    throw e;
   } else {
    interrupted = true;
   }
   }

   if (isDone) {
   return true;
   } else {
   waitTime = timeoutNanos - (System.nanoTime - startTime);
   if (waitTime <= 0) {
    return isDone;
   }
   }
  }
  }
 } finally {
  if (interrupted) {
  Thread.currentThread.interrupt;
  }
 }
 }

 @Override
 public IFuture<V> awaitUninterruptibly {
 try {
  return await0(false);
 } catch (InterruptedException e) { // 这里若抛异常了就无法处理了
  throw new java.lang.InternalError;
 }
 }
 
 @Override
 public boolean awaitUninterruptibly(long timeoutMillis) {
 try {
  return await0(TimeUnit.MILLISECONDS.toNanos(timeoutMillis), false);
 } catch (InterruptedException e) {
  throw new java.lang.InternalError;
 }
 }

 @Override
 public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
 try {
  return await0(unit.toNanos(timeout), false);
 } catch (InterruptedException e) {
  throw new java.lang.InternalError;
 }
 }

 protected IFuture<V> setFailure(Throwable cause) {
 if (setFailure0(cause)) {
  notifyListeners;
  return this;
 }
 throw new IllegalStateException("complete already: " + this);
 }

 private boolean setFailure0(Throwable cause) {
 if (isDone) {
  return false;
 }

 synchronized (this) {
  if (isDone) {
  return false;
  }
  result = new CauseHolder(cause);
  notifyAll;
 }

 return true;
 }

 protected IFuture<V> setSuccess(Object result) {
 if (setSuccess0(result)) { // 设置成功后通知监听器
  notifyListeners;
  return this;
 }
 throw new IllegalStateException("complete already: " + this);
 }

 private boolean setSuccess0(Object result) {
 if (isDone) {
  return false;
 }

 synchronized (this) {
  if (isDone) {
  return false;
  }
  if (result == null) { // 异步操作正常执行完毕的结果是null
  this.result = SUCCESS_SIGNAL;
  } else {
  this.result = result;
  }
  notifyAll;
 }
 return true;
 }

 private void notifyListeners {
 for (IFutureListener<V> l : listeners) {
  notifyListener(l);
 }
 }

 private void notifyListener(IFutureListener<V> l) {
 try {
  l.operationCompleted(this);
 } catch (Exception e) {
  e.printStackTrace;
 }
 }

 private static class SuccessSignal {

 }

 private static final class CauseHolder {
 final Throwable cause;

 CauseHolder(Throwable cause) {
  this.cause = cause;
 }
 }
} 

那么要怎么使用这个呢,有了上面的骨架实现,我们就可以定制各种各样的异步结果了。下面模拟一下一个延时的任务:

 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<Integer> {
  
  @Override
  public void operationCompleted(IFuture<Integer> 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<Integer> {
 
 @Override
 public IFuture<Integer> setSuccess(Object result) {
 return super.setSuccess(result);
 }
 
 @Override
 public IFuture<Integer> setFailure(Throwable cause) {
 return super.setFailure(cause);
 }
 
} 

可以看到客户端不用主动去询问future是否完成,而是future完成时自动回调operationcompleted方法,客户端只需在回调里实现逻辑即可。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。

 类似资料:
  • 本文向大家介绍Django异步任务线程池实现原理,包括了Django异步任务线程池实现原理的使用技巧和注意事项,需要的朋友参考一下 这篇文章主要介绍了Django异步任务线程池实现原理,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 当数据库数据量很大时(百万级),许多批量数据修改请求的响应会非常慢,一些不需要即时响应的任务可以放到后台的异步线程中

  • 前言 面试时,关于同步和异步,可能会问以下问题: 同步和异步的区别是什么?分别举一个同步和异步的例子 一个关于 setTimeout 的笔试题 前端使用异步的场景哪些? 面试时,关于js运行机制,需要注意以下几个问题: 如何理解JS的单线程 什么是任务队列 什么是 EventLoop 理解哪些语句会放入异步任务队列 理解语句放入异步任务队列的时机 JS的异步和单线程 因为是单线程,所以必须异步。

  • 本文向大家介绍Java多线程之多线程异常捕捉,包括了Java多线程之多线程异常捕捉的使用技巧和注意事项,需要的朋友参考一下   一:为什么要单独讲多线程的异常捕捉呢? 先看个例子:   在run中手动抛出了一个运行时异常,在main中启动线程,catch语句块中捕捉下异常,捕捉到打印一句话。运行结果如下图:   发现异常被抛到了控制台,没有打印catch块中的语句。   结论:多线程运行不能按照顺

  • 本文向大家介绍Java多线程 线程组原理及实例详解,包括了Java多线程 线程组原理及实例详解的使用技巧和注意事项,需要的朋友参考一下 线程组 线程组可以批量管理线程和线程组对象。 一级关联 例子如下,建立一级关联。 输出结果如下 每隔三秒输出两个线程名称,符合预期。 线程组自动归组属性 输出结果如下 没有指定线程组,则归属到当前线程所属的组。 根线程组 运行结果 当前线程的线程组的父线程组是系统

  • 我正在使用来处理通过文件的大量记录。每一行都是一条记录,我将每一行传递给单独的线程进行处理,问题是我必须收集这些处理过的记录以及在处理记录时生成的更多数据,然后在最后的数据收集上应用一些业务逻辑。我将一个通用的传递给所有线程来填充已处理的数据,当我通过visualVM调试它时,我发现(屏幕截图如下)这些线程在等待中花费的时间比在运行中花费的时间多。我想这是因为一个线程在写入时获得了锁。 有没有一种

  • 本文向大家介绍JAVA多线程之中断机制stop()、interrupted()、isInterrupted(),包括了JAVA多线程之中断机制stop()、interrupted()、isInterrupted()的使用技巧和注意事项,需要的朋友参考一下 一,介绍 本文记录JAVA多线程中的中断机制的一些知识点。主要是stop方法、interrupted()与isInterrupted()方法的区