scala.concurrent.Promise有什么用例？

Question

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

1. SIP说Depending on the implementation, it may be the case that p.future == p.怎么会这样？是Future和Promise不是两种不同的类型？

2. 我们Promise什么时候应该使用？示例producer and consumer代码:

   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的所有行为.

关于你的第一个问题,对于我们的承诺,怎么可能呢p.future == p？你可以想象这就像一个单项缓冲区 - 一个最初为空的容器,你可以在后面存储一个永远成为其内容的值.现在,根据你的观点,这既是承诺,也是未来.对于打算在缓冲区中写入值的人来说,这是有希望的.对于等待将该值放入缓冲区的人来说,这是一个未来.

具体来说,对于Scala并发API,如果您在此处查看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))

}

现在,可以在这里找到promises,DefaultPromise和KeptPromise的实现.它们都扩展了一个基本的小特征,恰好具有相同的名称,但它位于不同的包中:

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在这里:

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

因此,正如您所看到的,您从生产者块获得的结果会被投入到承诺中.

后期编辑:

关于现实世界的使用:大多数时候你不会直接处理承诺.如果您将使用执行异步计算的库,那么您将只使用库的方法返回的期货.在这种情况下,承诺是由图书馆创建的 - 您只是在阅读那些方法所做的阅读结束.

但是,如果您需要实现自己的异步API,则必须开始使用它们.假设您需要在Netty之上实现异步HTTP客户端.然后你的代码看起来会像这样

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