创建用于并行处理集合元素的 Akka 流

Question

我正在尝试为包含并行处理流的 Akka 流定义一个图（我正在使用 Akka.NET，但这应该没关系）。想象一个订单的数据源，每个订单由一个订单 ID 和一个产品列表（订单项）组成。工作流程如下：

1. 接收和订购
2. 将订单广播到两个流，流 A 处理订单商品，通道 B 处理订单 ID（一些记账工作）
3. 流程 A：将订单项目集合拆分为单独的元素，每个元素单独处理
4. 流程 A：对于上一步中拆分产生的每个订单商品，调用一些外部服务来查找额外信息（价格、可用性等）
5. 流程 B：为给定的订单 ID 做一些额外的记账
6. 合并流 A 和 B
7. 将上一步中的合并数据发送到接收器，从而丰富订单信息

步骤 1（Source.From）、2（广播）、4-5（地图）、6（合并）、7（接收器）看起来没问题。但是如何在 Akka 或响应式流术语中实现集合拆分？这不是广播或扁平化，需要将 N 个元素的集合拆分为 N 个独立的子流，这些子流稍后将被合并回来。这是如何实现的？

Answer 1

我建议一次性完成。我知道两个流程看起来更酷，但相信我，就设计的简单性而言，这是不值得的（我试过了）。你可以写这样的东西

import akka.stream.scaladsl.{Flow, Sink, Source, SubFlow}

import scala.collection.immutable
import scala.concurrent.Future

case class Item()

case class Order(items: List[Item])

val flow = Flow[Order]
  .mapAsync(4) { order =>
    Future {
      // Enrich your order here
      order
    }
  }
  .mapConcat { order =>
    order.items.map(order -> _)
  }
  .mapAsync(4) { case (order, item) =>
    Future {
      // Enrich your item here
      order -> item
    }
  }
  .groupBy(2, tuple => tuple._1)
  .fold[Map[Order, List[Item]]](immutable.Map.empty) { case (map, (order, item)) => map.updated(order, map.getOrElse(order, Nil) :+ item) }
  .mapConcat { _.map { case (order, newItems) => order.copy(items = newItems)} }

但即使是这种方法也很糟糕。上面的代码或您的设计可能会出现很多问题。如果某一订单商品的丰富化失败，您会怎么做？如果订单对象的丰富失败怎么办？您的信息流应该如何处理？

如果我是你，我会拥有 Flow[Order] 并在 mapAsync 中处理它的孩子，所以至少它可以保证我没有部分处理的订单。