好问题！在Akka Stream场景下，你完全可以实现自定义的溢出元素处理逻辑，而不是单纯依赖内置的OverflowStrategy。下面给你两种实用的方案，适配不同的代码结构场景：

方案一：封装队列统一处理溢出 #

这种方式适合多个地方需要向队列发送消息的场景，通过包装原始队列，统一在溢出时完成对应的Promise，避免重复代码。

首先定义自定义异常：

case class ThrottledException(message: String) extends Exception(message)

然后封装带溢出处理的队列创建逻辑：

import akka.stream.scaladsl.{Source, SourceQueueWithComplete}
import akka.stream.OverflowStrategy
import scala.concurrent.{Future, Promise}
import akka.stream.QueueOfferResult

def createProcessMessageQueue(bufferSize: Int): (
  Source[(ProcessMessageInputData, Promise[ProcessMessageOutputData]), _],
  Future[SourceQueueWithComplete[(ProcessMessageInputData, Promise[ProcessMessageOutputData])]]
) = {
  // 保留你原有的peekMatValue逻辑
  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)
  }

  val (source, rawQueueFuture) = peekMatValue(
    Source.queue[(ProcessMessageInputData, Promise[ProcessMessageOutputData])](bufferSize, OverflowStrategy.dropNew)
  )

  // 包装原始队列，重写offer方法处理溢出
  val wrappedQueueFuture = rawQueueFuture.map { rawQueue =>
    new SourceQueueWithComplete[(ProcessMessageInputData, Promise[ProcessMessageOutputData])] {
      override def offer(elem: (ProcessMessageInputData, Promise[ProcessMessageOutputData])): Future[QueueOfferResult] = {
        rawQueue.offer(elem).map {
          case QueueOfferResult.Dropped =>
            // 用自定义异常完成Promise
            elem._2.failure(ThrottledException("System is overloaded, request dropped due to throttling"))
            QueueOfferResult.Dropped
          case otherResult => otherResult
        }(scala.concurrent.ExecutionContext.global)
      }

      // 代理其他方法到原始队列
      override def complete(): Unit = rawQueue.complete()
      override def fail(ex: Throwable): Unit = rawQueue.fail(ex)
      override def watchCompletion(): Future[Unit] = rawQueue.watchCompletion()
    }
  }

  (source, wrappedQueueFuture)
}

方案说明 #

• 我们通过包装SourceQueueWithComplete，在offer方法返回Dropped结果时，主动触发对应Promise的失败逻辑。
• 原始队列的线程安全性完全保留，因为所有操作最终都会代理到原始队列。

方案二：在调用offer时直接处理溢出 #

如果只有少数地方需要向队列发送消息，这种方式更轻量，直接在调用offer的地方监听结果并处理：

// 在Web服务端点的业务逻辑中
import akka.stream.QueueOfferResult
import scala.concurrent.ExecutionContext.global

val inputData = ProcessMessageInputData(...)
val responsePromise = Promise[ProcessMessageOutputData]()

// 发送消息到队列并监听结果
queue.offer((inputData, responsePromise)).onComplete {
  case scala.util.Success(QueueOfferResult.Dropped) =>
    responsePromise.failure(ThrottledException("Request throttled: system is currently overloaded"))
  case _ => 
    // 其他情况（如Enqueued、QueueClosed）无需额外处理，由下游Sink完成Promise
}(global)

// 将Promise的Future返回给Web服务器作为响应
responsePromise.future

方案说明 #

• 这种方式不需要修改队列本身，直接在调用点处理溢出场景，代码更直观。
• 缺点是如果有多个调用点，需要重复编写处理逻辑。

注意事项 #

• 确保使用合适的ExecutionContext处理Promise的失败，避免阻塞IO线程。
• 如果你的队列是多线程并发调用的，两种方案都能保证线程安全，因为Akka的SourceQueueWithComplete本身就是线程安全的。

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