scala - scala.concurrent.Promise 有哪些用例？

Question

我正在阅读SIP-14 Future 的概念非常有意义且易于理解。但有两个关于 Promise 的问题:

1. SIP 表示根据实现的不同，可能会出现 p.future == p 的情况。怎么会这样？ Future 和 Promise 不是两种不同的类型吗？

2. 我们什么时候应该使用Promise？示例生产者和消费者代码:

   import scala.concurrent.{ future, promise }
   val p = promise[T]
   val f = p.future
   
   val producer = future {
       val r = produceSomething()
       p success r
       continueDoingSomethingUnrelated()
   }
   val consumer = future {
       startDoingSomething()
       f onSuccess {
           case r => doSomethingWithResult()
       }
   }
   

很容易阅读，但我们真的需要这样写吗？我尝试仅使用 Future 而没有 Promise 来实现它，如下所示:

val f = future {
   produceSomething()
}

val producer = future {
   continueDoingSomethingUnrelated()
}

startDoingSomething()

val consumer = future {
  f onSuccess {
    case r => doSomethingWithResult()
  }
}

这个例子和给定的例子有什么区别，是什么让 Promise 成为必要的？

Answer 1

promise 和 future 是互补的概念。 Future 是一个将在未来某个时间检索到的值，当该事件发生时您可以用它做一些事情。因此，它是计算的读取或输出端点 - 您可以从中检索值。

通过类比，Promise 是计算的写入端。您创建一个 Promise，在该 Promise 中放置计算结果，并从该 Promise 中获得一个 future，用于读取放入 Promise 中的结果。当您完成 Promise 时，无论失败还是成功，您都将触发附加到关联 Future 的所有行为。

关于你的第一个问题，对于promise p，我们怎么可能有p.future == p。您可以将其想象为一个单项缓冲区 - 一个最初为空的容器，之后您可以存储一个值，该值将永远成为其内容。现在，根据您的观点，这既是 promise 也是 future 。对于打算将值写入缓冲区的人来说，这是一个 promise 。对于等待该值放入缓冲区的人来说，这是一个 future 。

具体而言，对于 Scala 并发 API，如果您查看 here 中的 Promise 特征您可以看到 Promise 伴随对象中的方法是如何实现的:

object Promise {

  /** Creates a promise object which can be completed with a value.
   *  
   *  @tparam T       the type of the value in the promise
   *  @return         the newly created `Promise` object
   */
  def apply[T](): Promise[T] = new impl.Promise.DefaultPromise[T]()

  /** Creates an already completed Promise with the specified exception.
   *  
   *  @tparam T       the type of the value in the promise
   *  @return         the newly created `Promise` object
   */
  def failed[T](exception: Throwable): Promise[T] = new impl.Promise.KeptPromise[T](Failure(exception))

  /** Creates an already completed Promise with the specified result.
   *  
   *  @tparam T       the type of the value in the promise
   *  @return         the newly created `Promise` object
   */
  def successful[T](result: T): Promise[T] = new impl.Promise.KeptPromise[T](Success(result))

}

现在，Promise、DefaultPromise 和 KeptPromise 的实现可以在 here 找到。 。它们都扩展了一个基本的小特征，该特征恰好具有相同的名称，但它位于不同的包中:

private[concurrent] trait Promise[T] extends scala.concurrent.Promise[T] with scala.concurrent.Future[T] {
  def future: this.type = this
}

所以你可以明白 p.future == p 的含义。

DefaultPromise 是我上面提到的缓冲区，而 KeptPromise 是一个缓冲区，其值是从创建时就放入的。

关于您的示例，您在那里使用的 future block 实际上在幕后创建了一个 promise 。我们来看看here中future的定义:

def future[T](body: =>T)(implicit execctx: ExecutionContext): Future[T] = Future[T](body)

通过遵循一系列方法，您最终会得到 impl.Future :

private[concurrent] object Future {
  class PromiseCompletingRunnable[T](body: => T) extends Runnable {
    val promise = new Promise.DefaultPromise[T]()

    override def run() = {
      promise complete {
        try Success(body) catch { case NonFatal(e) => Failure(e) }
      }
    }
  }

  def apply[T](body: =>T)(implicit executor: ExecutionContext): scala.concurrent.Future[T] = {
    val runnable = new PromiseCompletingRunnable(body)
    executor.execute(runnable)
    runnable.promise.future
  }
}

因此，正如您所看到的，从生产者 block 获得的结果将被注入(inject)到 promise 中。

稍后编辑:

关于现实世界的使用:大多数时候你不会直接处理 promise 。如果您将使用执行异步计算的库，那么您只需使用该库方法返回的 futures 即可。在本例中，Promise 是由库创建的 - 您只需阅读这些方法的作用即可。

但是，如果您需要实现自己的异步 API，则必须开始使用它们。假设您需要在 Netty 之上实现一个异步 HTTP 客户端。那么你的代码将看起来有点像这样

    def makeHTTPCall(request: Request): Future[Response] = {
        val p = Promise[Response]
        registerOnCompleteCallback(buffer => {
            val response = makeResponse(buffer)
            p success response
        })
        p.future
    }