在Akka Streams中处理丢弃的消息

Question

我有以下源队列定义。

lazy val (processMessageSource, processMessageQueueFuture) =
   peekMatValue(
      Source
        .queue[(ProcessMessageInputData, Promise[ProcessMessageOutputData])](5, OverflowStrategy.dropNew))


def peekMatValue[T, M](src: Source[T, M]): (Source[T, M], Future[M])  {
    val p = Promise[M]
    val s = src.mapMaterializedValue { m =>
      p.trySuccess(m)
      m
    }  
    (s, p.future)
  }

进程消息输入数据类本质上是一个工件，它在调用者调用Web服务器端点时创建，该端点连接到此流（即服务端点的业务逻辑将消息放入此队列）。进程消息的Promise是在应用程序接收器的下游完成的，然后Web服务器对此未来进行完全回调以返回响应。

此流还有其他入口来源。

现在可以备份缓冲区，因为其他源可能使系统过载，从而触发流背压。现有代码只删除了新消息。但我仍然希望完成进程消息输出promise以完成一个异常，说明“Throttled”之类的内容。

是否有一种机制可以编写自定义溢出策略，或者对溢出元素进行后处理以允许我这样做？

Answer 1

根据https://github.com/akka/akka/blob/master/akkastream/src/main/scala/akka/stream/impl/QueueSource.scala#L83

dropNew会工作得很好。在客户端，它看起来像。

processMessageQueue.offer(in, pr).foreach { res =>
  res match {
    case Enqueued => // Code to handle case when successfully enqueued. 
    case Dropped => // Code to handle messages that are dropped since the buffier was overflowing. 
  }
}