阿卡流-如何从多个SQS源流

Question

这是 Akka 流 - 根据流中的元素选择接收器的后续帖子。

假设我有多个SQS队列，我想从这些队列中进行流式传输。我使用Alpakka的AWS SQS连接器来创建< code >源。

implicit val sqsClient: AmazonSQSAsync = ???
val queueUrls: List[String] = ???
val sources: List[Source[Message, NotUsed]] = queueUrls.map(url => SqsSource(url))

现在，我想将合并源代码以合并它们。但是，Source.combine方法不支持将列表作为参数传递，而只支持变量。

def combine[T, U](first: Source[T, _], second: Source[T, _], rest: Source[T, _]*)(strategy: Int ⇒ Graph[UniformFanInShape[T, U], NotUsed])

当然，我可以用手指输入所有的源参数。但是，如果我有10个源队列，这个参数会变得很长。

有没有一种方法可以将源列表中的源组合起来？

【补充】

正如拉蒙·J·罗梅罗·维吉尔（Ramon J Romero y Vigil）所指出的，保持“薄贴面”是更好的做法。然而，在这种特殊情况下，我使用单个sqsClient来初始化所有SqsSource。

Answer 1

您可以使用foldLeft连接或合并源：

val sources: List[Source[Message, NotUsed]] = ???

val concatenated: Source[Message, NotUsed] = sources.foldLeft(Source.empty[Message])(_ ++ _)
// the same as sources.foldLeft(Source.empty[Message])(_ concat _)

val merged: Source[Message, NotUsed] = sources.foldLeft(Source.empty[Message])(_ merge _)

或者，您可以将Source.zipN与flatMapConcat一起使用：

val combined: Source[Message, NotUsed] = Source.zipN(sources).flatMapConcat(Source.apply)