这段代码可以翻译成有状态的akka​​流吗？

Question

我正在尝试使用 akka 流来收听 sqs，我从它的 q 中获取消息 使用这个代码sn-p：

当然，这段代码 sn-p 一个接一个地获取消息（然后确认它）：

implicit val system = ActorSystem()
implicit val mat = ActorMaterializer()
implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(ioThreadPoolSize))
val awsSqsClient: AmazonSQSAsync = AmazonSQSAsyncClientBuilder
  .standard()
  .withCredentials(new ClasspathPropertiesFileCredentialsProvider())
  .withEndpointConfiguration(new EndpointConfiguration(sqsEndpoint, configuration.regionName))
  .build()

val future = SqsSource(sqsEndpoint)(awsSqsClient)
  .takeWhile(_ => true)
  .mapAsync(parallelism = 2)(m => {
    val msgBody = SqsMessage.deserializeJson(m.getBody)
    msgBody match {
      case Right(body) => val id = getId(body) //do some stuff with the message may save state according the id
    }
    Future(m, Ack())
  })
  .to(SqsAckSink(sqsEndpoint)(awsSqsClient))
  .run()

我的问题是： 我可以获取多条消息，并将它们保存在一个有状态的地图中以供以后使用吗？

例如收到 5 条消息后（所有消息都将保存（每个状态））

如果发生特定情况，我将全部确认，如果没有，它们将返回队列（因为可见性超时，无论如何都会发生）？

谢谢。

Answer 1

可能是您正在寻找grouped（或groupedWithin）组合子。这些允许您批量处理消息并分组处理它们。 groupedWithin 允许您在某个时间后释放一批，以防它尚未达到您确定的大小。文档参考here。

在随后的检查流程中，您可以执行所需的任何逻辑，并在您希望消息被确认时发出序列，否则不发出它们。

例子：

  val yourCheck: Flow[Seq[MessageActionPair], Seq[MessageActionPair], NotUsed] = ???

  val future = SqsSource(sqsEndpoint)(awsSqsClient)
    .takeWhile(_ => true)
    .mapAsync(parallelism = 2){ ... }
    .grouped(5)
    .via(yourCheck)
    .mapConcat(identity)
    .to(SqsAckSink(sqsEndpoint)(awsSqsClient))
    .run()