动态合并 Akka 流

我尝试使用 Akka 流通过以下方式构建 pub sub 总线：

发布者添加主题的源流，订阅者指定主题并获取该主题的所有内容。然而，主题可以由多个发布者发布，发布者和订阅者都可以随时加入。

我的想法是合并所有源，然后将过滤后的源返回给订阅者。

然而，由于发布者可以在任何点加入，因此在订阅之后可能会添加源，并且订阅者需要像该主题的任何其他已发布数据一样从中获取数据。

有没有一种方法可以动态管理流到源的合并，以便满足以下条件：

publish(topic: String, messages: Source[T])
subscribe(topic: String): Source[T]

这样，无论何时添加发布者，主题的订阅者都将在订阅后获取发布到与该主题相关的任何源的所有消息。

也很高兴听到替代方法。

谢谢， Z

你可能想看看这个 Akka 文档：构建一个dynamic pub-sub service https://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html?language=scala#combining-dynamic-operators-to-build-a-simple-publish-subscribe-service using MergeHub https://doc.akka.io/api/akka/2.5/akka/stream/scaladsl/MergeHub%24.html and BroadcastHub https://doc.akka.io/api/akka/2.5/akka/stream/scaladsl/BroadcastHub%24.html.

这是使用的示例代码MergeHub https://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html?language=scala#using-the-mergehub and a BroadcastHub https://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html?language=scala#using-the-broadcasthub分别作为动态扇入和扇出结点。

这个想法是连接一个MergeHub with a BroadcastHub以 Flow via 的形式形成一个 pub-sub 通道Flow.fromSinkAndSource https://doc.akka.io/docs/akka/2.5/stream/operators/Flow/fromSinkAndSource.html:

val (bfSink, bfSource) = MergeHub.source[String](perProducerBufferSize).
  toMat(BroadcastHub.sink[String](bufferSize))(Keep.both).
  run

val busFlow: Flow[String, String, NotUsed] = Flow.fromSinkAndSource(bfSink, bfSource)

注意Keep.both上面的代码片段生成一个物化值的元组(Sink[T, NotUsed], Source[T, NotUsed]) from MergeHub.source[T] and BroadcastHub.sink[T]其具有以下方法签名：

object MergeHub {
  def source[T](perProducerBufferSize: Int): Source[T, Sink[T, NotUsed]] = // ...
  // ...
}

object BroadcastHub {
  def sink[T](bufferSize: Int): Sink[T, Source[T, NotUsed]] = // ...
  // ...
}

下面是一个简单的发布-订阅通道的示例代码busFlow（类似于 Akka 文档中的示例）：

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
import akka.NotUsed

implicit val system = ActorSystem("system")
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher

val (bfSink, bfSource) = MergeHub.source[String](perProducerBufferSize = 32).
  toMat(BroadcastHub.sink[String](bufferSize = 256))(Keep.both).
  run

// Optional: avoid building up backpressure when there is no subscribers
bfSource.runWith(Sink.ignore)

val busFlow: Flow[String, String, NotUsed] = Flow.fromSinkAndSource(bfSink, bfSource)

Testing busFlow:

Source(101 to 103).map(i => s"Batch(A)-$i").
  delay(2.seconds, DelayOverflowStrategy.backpressure).
  viaMat(busFlow)(Keep.right).
  to(Sink.foreach{ case s: String => println("Consumer(1)-" + s) }).
  run

Source(104 to 105).map(i => s"Batch(B)-$i").
  viaMat(busFlow)(Keep.right).
  to(Sink.foreach{ case s: String => println("Consumer(2)-" + s) }).
  run

// Consumer(2)-Batch(B)-104
// Consumer(2)-Batch(B)-105
// Consumer(1)-Batch(B)-104
// Consumer(1)-Batch(B)-105
// Consumer(1)-Batch(A)-101
// Consumer(1)-Batch(A)-102
// Consumer(2)-Batch(A)-101
// Consumer(1)-Batch(A)-103
// Consumer(2)-Batch(A)-102
// Consumer(2)-Batch(A)-103

作为一个 pub-sub 通道，输入busFlow发布于bfSink发送给所有订阅者，同时其输出流通过bfSource已发布的所有元素。例如：

val p1 = Source.tick[Int](0.seconds, 5.seconds, 5).map(_.toString)
p1.runWith(bfSink)

val p2 = Source.tick[Int](2.seconds, 10.seconds, 10).map(_.toString)
p2.runWith(bfSink)

val s1 = bfSource
s1.runForeach(x => println(s"s1 --> $x"))

val s2 = bfSource
s2.runForeach(x => println(s"s2 --> $x"))

// s1 --> 5
// s2 --> 5
// s1 --> 10
// s2 --> 10
// s2 --> 5
// s1 --> 5
// s2 --> 5
// s1 --> 5
// s1 --> 10
// s2 --> 10
// s2 --> 5
// s1 --> 5
// ...

其他可能感兴趣的相关主题包括KillSwitch https://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html?language=scala#controlling-stream-completion-with-killswitch用于流完成控制和PartitionHub https://doc.akka.io/docs/akka/2.5/stream/stream-dynamic.html?language=scala#using-the-partitionhub用于将 Stream 元素从给定的生产者路由到一组动态的消费者。