现在我正在使用
AmqpSource.atMostOnceSource(
NamedQueueSourceSettings(..))
我希望能够订阅多个交换。谁能帮我这个?
如果对于特定的 alpakka 源没有特定的内容,您可以使用 Merge 或 MerbeHub。
如果您事先知道所有来源,则可以使用合并将多个来源合并为一个
如果您事先不知道所有来源,您可以使用MergeHub,例如
// A simple consumer that will print to the console for now
val consumer = Sink.foreach(println)
// Attach a MergeHub Source to the consumer. This will materialize to a
// corresponding Sink.
val runnableGraph: RunnableGraph[Sink[String, NotUsed]] =
MergeHub.source[String](perProducerBufferSize = 16).to(consumer)
// By running/materializing the consumer we get back a Sink, and hence
// now have access to feed elements into it. This Sink can be materialized
// any number of times, and every element that enters the Sink will
// be consumed by our consumer.
val toConsumer: Sink[String, NotUsed] = runnableGraph.run()
// Feeding two independent sources into the hub.
AmqpSource.atMostOnceSource(
NamedQueueSourceSettings(..)).runWith(toConsumer)
AmqpSource.atMostOnceSource(
NamedQueueSourceSettings(..)).runWith(toConsumer)