Akka Stream：如何从多个SQS源进行流处理

嘿，这个场景我太熟悉了！面对一堆SQS队列要合并成一个流，还要避免手动写10个参数的麻烦，其实用Scala的语法糖就能轻松搞定，下面给你两种实用的方案：

方案一：用: _*将列表转为可变参数 #

Source.combine的签名确实要求第一个参数是单个Source，后面跟着可变参数列表，但我们可以把已有的Source列表拆成「第一个元素」+「剩余元素的可变参数」，借助Scala的: _*语法糖把剩余列表展开：

步骤1：先创建所有SQS Source的列表 #

假设你已经有10个SQS队列的URL，先把对应的Source都放到一个列表里：

import akka.stream.alpakka.sqs.scaladsl.SqsSource
import akka.stream.scaladsl.Source
import software.amazon.awssdk.services.sqs.model.Message

// 这里替换成你的10个队列URL
val queueUrls = List("url1", "url2", "url3", /* ... 剩下7个 */)
val sqsSources: List[Source[Message, NotUsed]] = queueUrls.map(SqsSource(_))

步骤2：用Source.combine合并 #

根据你的流处理需求选择合并策略：

• 如果要并行合并所有源的元素（元素会按到达顺序混合）：用Merge策略
• 如果要按顺序逐个消费每个源（先消费完第一个源的所有元素，再开始第二个）：用Concat策略

示例代码（以Merge为例）：

import akka.stream.scaladsl.Merge

val combinedSource = sqsSources match {
  case head :: tail => Source.combine(head, tail: _*)(Merge(_))
  case Nil => Source.empty // 处理空列表的边界情况
}

这里的: _*会把tail这个List展开成Source.combine需要的可变参数，完美适配它的签名，再也不用手动写10个参数啦！

方案二：用Graph DSL手动构建合并逻辑 #

如果你需要更灵活的合并控制（比如自定义合并规则），可以直接用Akka Stream的Graph DSL来构建：

import akka.stream.scaladsl.GraphDSL
import akka.stream.Graph
import akka.stream.UniformFanInShape

val mergeGraph: Graph[UniformFanInShape[Message, Message], NotUsed] = GraphDSL.create() { implicit builder =>
  import GraphDSL.Implicits._
  val merge = builder.add(Merge[Message](sqsSources.size))
  sqsSources.foreach(source => source ~> merge.in)
  UniformFanInShape(merge.out, merge.inSeq: _*)
}

val combinedSource = Source.fromGraph(mergeGraph)

这种方式适合复杂场景，但如果只是简单合并，方案一已经足够简洁。

内容的提问来源于stack exchange，提问作者gyoho