通过 akka 流链接上下文

Question

我正在将一些 C# 代码转换为 scala 和 akka 流。

我的 c# 代码如下所示：

Task<Result1> GetPartialResult1Async(Request request) ...
Task<Result2> GetPartialResult2Async(Request request) ...

async Task<Result> GetResultAsync(Request request) 
{
    var result1 = await GetPartialResult1Async(request);
    var result2 = await GetPartialResult2Async(request);
    return new Result(request, result1, result2);
}

现在是 akka 流。我没有从 Request 到结果的 Task 的函数，而是从请求流向结果。

所以我已经有了以下两个流程：

val partialResult1Flow: Flow[Request, Result1, NotUsed] = ...
val partialResult2Flow: Flow[Request, Result2, NotUsed] = ...

但是我看不出如何将它们组合成一个完整的流程，因为在第一个流程上调用 via 我们会丢失原始请求，而通过在第二个流程上调用 via 我们会丢失第一个流程的结果。

所以我创建了一个看起来像这样的 WithState monad：

case class WithState[+TState, +TValue](value: TValue, state: TState) {
  def map[TResult](func: TValue => TResult): WithState[TState, TResult] = {
    WithState(func(value), state)
  }
  ... bunch more helper functions go here
}

然后我将我的原始流程重写为如下所示：

def partialResult1Flow[TState]: Flow[WithState[TState, Request], WithState[TState, Result1]] = ...
def partialResult2Flow: Flow[WithState[TState, Request], WithState[TState, Result2]] = ...

并像这样使用它们：

val flow = Flow[Request]
    .map(x => WithState(x, x))
    .via(partialResult1Flow)
    .map(x => WithState(x.state, (x.state, x.value))
    .via(partialResult2Flow)
    .map(x => Result(x.state._1, x.state._2, x.value))

现在这可行，但我当然不能保证如何使用 flow。所以我真的应该让它接受一个状态参数：

def flow[TState] = Flow[WithState[TState, Request]]
    .map(x => WithState(x.value, (x.state, x.value)))
    .via(partialResult1Flow)
    .map(x => WithState(x.state._2, (x.state, x.value))
    .via(partialResult2Flow)
    .map(x => WithState(Result(x.state._1._2, x.state._2, x.value), x.state._1._1))

现在在这个阶段，我的代码变得非常难以阅读。我可以通过命名函数来清理它，并使用案例类而不是元组等，但基本上这里有很多附带的复杂性，这是很难避免的。

我错过了什么吗？这不是 Akka 流的好用例吗？有一些内置的方法吗？

Answer 1

我没有任何根本不同的方式来做到这一点，而不是我在问题中描述的。

但是电流流量可以显着改善：

第一阶段：FlowWithContext

可以使用内置的FlowWithContext，而不是使用自定义的WithState monad。

这样做的好处是您可以在流上使用标准运算符，而无需担心转换 WithState monad。 Akka 会为您解决这个问题。

所以不是

def partialResult1Flow[TState]: Flow[WithState[TState, Request], WithState[TState, Result1]] = 
    Flow[WithState[TState, Request]].mapAsync(_ mapAsync {doRequest(_)})

我们可以这样写：

def partialResult1Flow[TState]: FlowWithContext[Request, TState, Result1, TState, NotUsed] = 
    FlowWithContext[Request, TState].mapAsync(doRequest(_))

不幸的是，虽然 FlowWithContext 在您不需要更改上下文时很容易编写，但当您需要通过需要将一些当前数据移动到上下文（就像我们的那样）。为此，您需要转换为Flow（使用asFlow），然后使用asFlowWithContext 转换回FlowWithContext。

我发现在这种情况下，将整个内容写成 Flow 并在最后转换为 FlowWithContext 是最简单的方法。

例如：

def flow[TState]: FlowWithContext[Request, TState, Result, TState, NotUsed] = 
  Flow[(Request, TState)]
    .map(x => (x._1, (x._1, x._2)))
    .via(partialResult1Flow)
    .map(x => (x._2._1, (x._2._1, x._1, x._2._2))
    .via(partialResult2Flow)
    .map(x => (Result(x._2._1, x._2._2, x._1), x._2._2))
    .asFlowWithContext((a: Request, b: TState) => (a,b))(_._2)
    .map(_._1)

这样更好吗？

在这种特殊情况下，情况可能更糟。在其他情况下，您很少需要更改上下文，它会更好。但是无论哪种方式，我都建议使用它，因为它是内置的，而不是依赖于自定义 monad。

第 2 阶段：通过使用

为了使这更加用户友好，我为 Flow 和 FlowWithContext 创建了一个 viaUsing 扩展方法：

import akka.stream.{FlowShape, Graph}
import akka.stream.scaladsl.{Flow, FlowWithContext}

object FlowExtensions {
  implicit class FlowViaUsingOps[In, Out, Mat](val f: Flow[In, Out, Mat]) extends AnyVal {
    def viaUsing[Out2, Using, Mat2](func: Out => Using)(flow: Graph[FlowShape[(Using, Out), (Out2, Out)], Mat2]) : Flow[In, (Out2, Out), Mat] =
      f.map(x => (func(x), x)).via(flow)
  }

  implicit class FlowWithContextViaUsingOps[In, CtxIn, Out, CtxOut, Mat](val f: FlowWithContext[In, CtxIn, Out, CtxOut, Mat]) extends AnyVal {
    def viaUsing[Out2, Using, Mat2](func: Out => Using)(flow: Graph[FlowShape[(Using, (Out, CtxOut)), (Out2, (Out, CtxOut))], Mat2]):
    FlowWithContext[In, CtxIn, (Out2, Out), CtxOut, Mat] =
      f
        .asFlow
        .map(x => (func(x._1), (x._1, x._2)))
        .via(flow)
        .asFlowWithContext((a: In, b: CtxIn) => (a,b))(_._2._2)
        .map(x => (x._1, x._2._1))
  }
}

viaUsing 的目的是根据当前输出为FlowWithContext 创建输入，同时通过上下文传递它来保留当前输出。它产生一个Flow，其输出是嵌套流和原始流的输出的元组。

使用viaUsing，我们的示例简化为：

  def flow[TState]: FlowWithContext[Request, TState, Result, TState, NotUsed] =
    FlowWithContext[Request, TState]
      .viaUsing(x => x)(partialResult1Flow)
      .viaUsing(x => x._2)(partialResult2Flow)
      .map(x => Result(x._2._2, x._2._1, x._1))

我认为这是一项重大改进。我已请求将 viaUsing 添加到 Akka，而不是依赖扩展方法 here。

Answer 2

我同意使用 Akka Streams 进行背压很有用。但是，我不相信将partialResults 的计算建模为流在这里有用。拥有基于Futures 的“内部”逻辑并将其包装在流程的mapAsync 中，以将背压作为一个单元应用于整个操作，这似乎更简单，甚至可能更好。

这基本上是对 Levi Ramsey 早期出色答案的简化重构：

import scala.concurrent.{ ExecutionContext, Future }
import akka.NotUsed
import akka.stream._
import akka.stream.scaladsl._

case class Request()
case class Result1()
case class Result2()
case class Response(r: Request, r1: Result1, r2: Result2)

def partialResult1(req: Request): Future[Result1] = ???
def partialResult2(req: Request): Future[Result2] = ???

val system = akka.actor.ActorSystem()
implicit val ec: ExecutionContext = system.dispatcher

val flow: Flow[Request, Response, NotUsed] =
  Flow[Request]
    .mapAsync(parallelism = 12) { req =>
      for {
        res1 <- partialResult1(req)
        res2 <- partialResult2(req)
      } yield (Response(req, res1, res2))
    }

我会从这个开始，只有当你知道你有理由将partialResult1 和partialResult2 分成不同的阶段时，才会在Flow 中引入一个中间步骤。根据您的要求mapAsyncUnordered 可能更合适。

Answer 3

免责声明，我对 C# 的 async/await 并不完全熟悉。

根据我通过快速阅读 C# 文档所收集到的信息，Task<T> 是一个严格（即急切，而不是懒惰）评估的计算，如果成功最终将包含 T。 Scala 等价于 Future[T]，其中 C# 代码等价于：

import scala.concurrent.{ ExecutionContext, Future }

def getPartialResult1Async(req: Request): Future[Result1] = ???
def getPartialResult2Async(req: Request): Future[Result2] = ???

def getResultAsync(req: Request)(implicit ectx: ExecutionContext): Future[Result] = {
  val result1 = getPartialResult1Async(req)
  val result2 = getPartialResult2Async(req)
  result1.zipWith(result2) { tup => val (r1, r2) = tup
    new Result(req, r1, r2)
  }
  /* Could also:
   *   for {
   *     r1 <- result1
   *     r2 <- result2
   *    } yield { new Result(req, r1, r2) }
   *
   * Note that both the `result1.zipWith(result2)` and the above `for`
   * construction may compute the two partial results simultaneously.  If you
   * want to ensure that the second partial result is computed after the first 
   * partial result is successfully computed:
   *   for {
   *     r1 <- getPartialResult1Async(req)
   *     r2 <- getPartialResult2Async(req)
   *   } yield new Result(req, r1, r2)
   */
}

这种特殊情况不需要 Akka Streams，但如果您有其他需要使用 Akka Streams，您可以将其表示为

val actorSystem = ??? // In Akka Streams 2.6, you'd probably have this as an implicit val
val parallelism = ??? // Controls requests in flight

val flow = Flow[Request]
  .mapAsync(parallelism) { req =>
    import actorSystem.dispatcher

    getPartialResult1Async(req).map { r1 => (req, r1) }
  }
  .mapAsync(parallelism) { tup =>
    import actorSystem.dispatcher

    getPartialResult2Async(tup._2).map { r2 =>
      new Result(tup._1, tup._2, r2)
    }
  }

  /* Given the `getResultAsync` function in the previous snippet, you could also:
   *   val flow = Flow[Request].mapAsync(parallelism) { req =>
   *     getResultAsync(req)(actorSystem.dispatcher)
   *   }
   */

基于Future 的实现的一个优点是它很容易与您想在给定上下文（例如猫、akka 流、akka）中使用的任何 Scala 并发/并行抽象抽象集成。我对 Akka Streams 集成的一般直觉是我在第二个代码块中的评论中的三线方向。