Akka Stream 动态 Sink 取决于来自 Kafka 主题的消息

Question

我有一个阅读 Message 的 Kafka 消费者。每条消息都有一个 ID 和内容。

case class Message(id: String, content: String)

根据 ID，我想将消息写入单独的接收器。具体到一个 MongoDB 集合中。 Mongo 提供了一个 Sink 将其写入 DB 到指定的集合中。

val sink: Sink[Document, Future[Done]] = MongoSink.insertOne(collection(id))

问题是，我需要在连接 Kafka 消费者源时指定接收器，但每个元素都定义了它应该进入哪个接收器。 有没有办法在元素到达时动态使用特定的接收器。或者这是不可能的，例如，我应该为每个 id 使用不同的 Kafka 主题并将每个源连接到单独的接收器？

Answer 1

您的示例中的类型如何排列并不完全清楚（例如 Document 和 Message 之间的关系），但您可以采取以下几种方法：

• 如果有很多可能的集合并且无法提前知道它们，那么 Akka Streams 中最不坏的选择将是沿线

Sink.foreachAsync[Message](parallelism) { msg =>
  val document = documentFromMessage(msg)
  val collection = collection(msg.id)
  Source.single(document).runWith(MongoSink.insertOne(collection))
}

请注意，这将为每条消息使用一个新的 Mongo 接收器，这可能会带来效率问题。请注意，如果有一种更轻量级的方式（例如在 reactivemongo 驱动程序中？）在插入单个文档后返回 Future，但使用连接池之类的东西来减少单文档插入的开销，那可能会更可取。

• 如果预先知道集合，您可以为每个集合预构建接收器并使用Partition 和GraphDSL 定义一个包含预构建接收器的接收器

// collection0, etc. are predefined and encompass all of the collections which might be returned by collection(id)
val collections: Map[MongoCollection[Document], (Int, Sink[Document, Future[Done]])] = Map(
  collection0 -> (0 -> MongoSink.insertOne(collection0)),
  collection1 -> (1 -> MongoSink.insertOne(collection1)),
  collection2 -> (2 -> MongoSink.insertOne(collection2)),
  collection3 -> (3 -> MongoSink.insertOne(collection3))
)

val combinedSink = Sink.fromGraph(GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._

  val partition = builder.add(
    Partition[Message](
      collections.size,
      { msg => collections(collection(msg.id))._1 }
    )
  )

  val toDocument = Flow[Message].map(documentFromMessage)

  collections.foreach {
    case (_, (n, sink)) =>
      partition.out(n) ~> toDocument ~> sink
  }

  SinkShape.of(partition.in)
}