IT博客汇 | [转][转]并发编程 Promise, Future 和 Callback

[转][转]并发编程 Promise, Future 和 Callback

heiyeshuwu发表于 2017-03-07 16:01:33

在并发编程中，我们通常会用到一组非阻塞的模型：Promise，Future 和 Callback。其中的 Future 表示一个可能还没有实际完成的异步任务的结果，针对这个结果可以添加 Callback 以便在任务执行成功或失败后做出对应的操作，而 Promise 交由任务执行者，任务执行者通过 Promise 可以标记任务完成或者失败。可以说这一套模型是很多异步非阻塞架构的基础。

这一套经典的模型在 Scala、C# 中得到了原生的支持，但 JDK 中暂时还只有无 Callback 的 Future 出现，当然也并非在 JAVA 界就没有发展了，比如 Guava 就提供了ListenableFuture 接口，而 Netty 4+ 更是提供了完整的 Promise、Future 和 Listener 机制，在 Netty 的官方文档 Using as a generic library 中也介绍了将 Netty 作为一个 lib 包依赖，并且使用 Listenable futures 的示例。在实际的项目使用中，发现 Netty 的 EventLoop 机制不一定适用其他场景，因此想去除对 EventLoop 的依赖，实现一个简化版本。

参考 Scala 和 Netty 的代码重新定义了接口和实现，先介绍下和 Netty 版本的区别:

去除了对 EventLoop 的依赖，Callback 的执行策略不同：任务未完成时添加的 Callback，会在结束任务的线程执行；任务完成后添加的 Callback 会在添加 Callback 线程立即执行
一个 Callback 执行后会立即被清理
Callback 可以根据任务结果添加，支持添加以下三种 Callback: onComplete, onSuccess, onFailure, 不需要和 Netty 的 FutureListener 一样大部分场景下都需要检查 future.isSuccess 等
支持 Callback 的组合，Callback 包含一些函数式的方法，比如 compose 和 andThen 可以用来组合
使用 CountdownLatch 替换掉了 Netty 的 wait/notify 实现
去掉 Netty Future 一些不常使用的方法，同时补充一些模型间关联的方法，比如 Promise.getFuture

然后再介绍几个使用这个 commons-future 的示例:

异步执行任务，获得 Future 后添加 Callback
01 final TaskPromise promise = new DefaultTaskPromise();
02 final TaskFuture future = promise.getFuture();
03 final CountDownLatch latch = new CountDownLatch(1);
04 future.onComplete(new TaskCallback() { // 添加结束 Callback
05     @Override
06     public TaskFuture apply(TaskFuture f) {
07         latch.countDown();
08         return f;
09     }
10 });
11 new Thread(new Runnable() {
12     @Override
13     public void run() {
14         promise.setSuccess(null);
15     }
16 }).start();
17 latch.await();
异步执行任务，获得 Future 后添加成功结束的 Callback
01 final TaskPromise promise = new DefaultTaskPromise();
02 final TaskFuture future = promise.getFuture();
03 final CountDownLatch latch = new CountDownLatch(1);
04 future.onSuccess(new TaskCallback() { // 添加成功结束 Callback
05     @Override
06     public TaskFuture apply(TaskFuture f) {
07         latch.countDown();
08         return f;
09     }
10 });
11 new Thread(new Runnable() {
12     @Override
13     public void run() {
14         promise.setSuccess(null);
15     }
16 }).start();
17 latch.await();
异步执行任务，获得 Future 后，添加失败结束的组合 Callback
01 final TaskPromise promise = new DefaultTaskPromise();
02 final TaskFuture future = promise.getFuture();
03 final CountDownLatch latch = new CountDownLatch(2);
04 future.onFailure(new TaskCallback() {
05     @Override
06     public TaskFuture apply(TaskFuture f) {
07         latch.countDown();
08         return f;
09     }
10 }.andThen(new TaskCallback() {
11     @Override
12     public TaskFuture apply(TaskFuture f2) {
13         latch.countDown();
14         return f2;
15     }
16 }));
17 new Thread(new Runnable() {
18     @Override
19     public void run() {
20         promise.setFailure(new IllegalStateException("cm"));
21     }
22 }).start();
23 latch.await();
异步执行任务，获得 Future 后阻塞等待任务完成
01 final TaskPromise promise = new DefaultTaskPromise();
02 final TaskFuture future = promise.getFuture();
03 new Thread(new Runnable() {
04     @Override
05     public void run() {
06         try {
07             TimeUnit.SECONDS.sleep(2);
08         } catch (InterruptedException e) {
09         }
10         promise.setFailure(new IllegalStateException("cm"));
11     }
12 }).start();
13 future.await();

代码仓库: https://bitbucket.org/qiyi/commons-future

源文链接: http://isouth.org/archives/354.html

参考:

来源：http://isouth.org/archives/354.html

Scala的futue和promise

scala的future和promise

异步操作的有两个经典接口：Future和Promise,其中的 Future 表示一个可能还没有实际完成的异步任务的结果，针对这个结果可以添加 Callback 以便在任务执行成功或失败后做出对应的操作，而 Promise 交由任务执行者，任务执行者通过 Promise 可以标记任务完成或者失败。可以说这一套模型是很多异步非阻塞架构的基础。

scala的Future表示一个异步操作的结果状态，它持有一个值，在未来的某个之间点可用，该值是异步操作的结果,当异步操作没有完成，那么Future的isCompleted为false,当异步操作完成了且返回了值，那么Future的isCompleted返回true且是success, 如果异步操作没有完成或者异常终止，那么Future的isCompleted也返回true但是是failed.

scala的Future有一个重要特性，就是他只能被赋值一次，一旦Future对象被赋值了或者设置为异常，那么它会变成不可变的

创建一个Future最简单的方式就是调用future函数：它会启动一个异步操作且返回一个包含了操作结果的。一个future函数完成该值就可以用了，Future[T]的泛型是异步操作结果的类型。下面是一个创建Future的例子。

package com.joy.lession1
import scala.concurrent._
import ExecutionContext.Implicits.global
object FutureTest  extends App{
  val s = "Hello"
  val f: Future[String] = future {
    s + " future!"
  }
  f onSuccess {
    case msg => println(msg)
  }
  println(s)  //不加这句, f onSuccess就不执行
}

Future的apply()方法会构建一个异步操作且在未来某一个时刻返回一个值。Future的apply()方法内部使用了Promise。

下面给Future添加一个回调函数。

import scala.concurrent._
import ExecutionContext.Implicits.global
import scala.util.{Failure, Success, Try}
object FutureTest  extends App{
  val s = "Hello"
  val f: Future[String] = future {
    s + " future!"
  }
    f onComplete {
      case Success(t) =>
      {
        println(t)
      }
      case Failure(e) =>
      {
        println(s"An error has occured: $e.getMessage")
      }
    }
    println("dd")
}

Future的值不知道什么时候可用，所以需要一种机制来获取异步操作的结果，一种是不停的查看Future的完成状态，另一个采用阻塞的方式，scala提供了第二种方式的支持，使用scala.concurrent.Await,它有两个方法，一个是Await.ready，当Future的状态为完成时返回，一种是Await.result，直接返回Future持有的结果。Future还提供了一些map,filter,foreach等操作

Scala的Promise可以被认为一个是可写的，静态单赋值，它可以创建一个Future和完成一个Future(success complete和failed complete） promise字面意思是承诺，以为他可以控制异步操作的结果

import java.util.Timer
import java.util.TimerTask
import scala.concurrent._
object TimedEvent {
  val timer = new Timer
  /** Return a Future which completes successfully with the supplied value after secs seconds. */
  def delayedSuccess[T](secs: Int, value: T): Future[T] = {
    val result = Promise[T]
    timer.schedule(new TimerTask() {
      def run() = {
        result.success(value)
      }
    }, secs * 1000)
    result.future
  }
  /** Return a Future which completes failing with an IllegalArgumentException after secs
    * seconds. */
  def delayedFailure(secs: Int, msg: String): Future[Int] = {
    val result = Promise[Int]
    timer.schedule(new TimerTask() {
      def run() = {
        result.failure(new IllegalArgumentException(msg))
      }
    }, secs * 1000)
    result.future
  }

作者：半兽人
链接：http://www.orchome.com/250