Akka Http复用物化流处理POST请求：如何从Sink获取响应？

我完全理解你的需求：不想为每个HTTP请求重新物化整个流，而是复用一个预先启动的流来处理所有请求，同时还要能将流的处理结果返回给对应的客户端。下面是几种适合这个场景的实现方案，从简洁到灵活都有覆盖：

方案1：使用Source.queue（推荐，最简洁） #

Source.queue是Akka Stream专门为“外部向流推送元素”的场景设计的API，它允许你预先物化一个流，然后通过队列接口将请求发送进去，同时可以轻松关联每个请求的响应回调。

步骤1：预先物化共享流 #

首先定义一个封装请求和响应Promise的类，然后创建并启动共享流：

import akka.stream.scaladsl.{Sink, Source, Flow}
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import scala.concurrent.Promise

// 封装请求和对应的响应Promise，用来关联请求和结果
case class QueuedRequest(req: HttpRequest, responsePromise: Promise[HttpResponse])

// 假设这是你的处理Flow链（替换成你实际的多个Flow）
val processingFlow: Flow[QueuedRequest, Unit, NotUsed] = Flow[QueuedRequest].map { queued =>
  // 这里写你的请求处理逻辑，比如解析请求体、执行业务操作等
  val response = HttpResponse(200, entity = s"Processed request: ${queued.req.uri}")
  
  // 将结果写入Promise，Akka HTTP会自动处理后续的响应返回
  queued.responsePromise.success(response)
}

// 物化共享流，得到队列接口和流的完成Future
val (requestQueue, streamCompletion) = Source.queue[QueuedRequest](
  bufferSize = 100, // 根据你的并发需求调整缓冲区大小
  overflowStrategy = OverflowStrategy.dropNew // 缓冲区满时的处理策略，可按需调整
)
.via(processingFlow)
.to(Sink.ignore) // 因为我们已经通过Promise返回结果，Sink可以忽略输出
.run() // 这里只物化一次，流会持续运行

步骤2：在Akka HTTP路由中使用队列 #

在路由里，每个请求创建一个Promise，将请求放入队列，并通过Promise获取响应：

import akka.http.scaladsl.server.Directives._
import scala.concurrent.ExecutionContext.Implicits.global
import akka.stream.QueueOfferResult

val route: Route = path("post" / Segment) { p =>
  post {
    extractRequest { req =>
      val responsePromise = Promise[HttpResponse]()
      
      // 将请求放入共享队列
      requestQueue.offer(QueuedRequest(req, responsePromise)).onComplete {
        case Success(QueueOfferResult.Enqueued) =>
          // 请求已入队，等待流处理
        case Success(QueueOfferResult.Dropped) =>
          responsePromise.failure(new RuntimeException("请求被丢弃：流缓冲区已满"))
        case Success(QueueOfferResult.Failure(ex)) =>
          responsePromise.failure(new RuntimeException("流处理失败", ex))
        case Success(QueueOfferResult.QueueClosed) =>
          responsePromise.failure(new RuntimeException("流已关闭"))
        case Failure(ex) =>
          responsePromise.failure(ex)
      }
      
      // 将Promise转为Future，Akka HTTP会自动完成响应
      complete(responsePromise.future)
    }
  }
}

方案2：使用Source.actorRef（更灵活，适合复杂场景） #

如果你的流处理逻辑需要和Actor系统深度集成，或者需要更灵活的消息控制，可以用Source.actorRef创建流的源，通过Actor接收请求，再将结果返回给请求方。

步骤1：定义消息类型和物化共享流 #

首先定义携带请求和回复通道的消息，然后创建并启动流：

import akka.actor.{Actor, ActorRef, Props}
import akka.stream.scaladsl.{Sink, Source, Flow}
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}

// 定义消息：携带请求和回复用的ActorRef
case class ProcessRequest(req: HttpRequest, replyTo: ActorRef[HttpResponse])

// 你的处理Flow链（这里将ProcessRequest转换为带回复的结果）
val processingFlow: Flow[ProcessRequest, Unit, NotUsed] = Flow[ProcessRequest].map { msg =>
  // 处理请求得到响应
  val response = HttpResponse(200, entity = s"Processed request with path param: ${msg.req.uri.path}")
  
  // 将响应发送回请求方的Actor
  msg.replyTo ! response
}

// 物化流，得到源Actor的引用
val (streamSourceActor, streamCompletion) = Source.actorRef[ProcessRequest](
  completionMatcher = PartialFunction.empty, // 这里不需要特殊的完成消息
  failureMatcher = PartialFunction.empty,
  bufferSize = 100,
  overflowStrategy = OverflowStrategy.dropNew
)
.via(processingFlow)
.to(Sink.ignore)
.run()

步骤2：在路由中使用Actor发送请求并接收响应 #

在路由里，为每个请求创建一个临时Actor来接收响应，然后将请求发送给流的源Actor：

import akka.http.scaladsl.server.Directives._
import scala.concurrent.{Promise, ExecutionContext}
import akka.pattern.ask
import akka.util.Timeout

val route: Route = path("post" / Segment) { p =>
  post {
    extractRequest { req =>
      extractMaterializer { mat =>
        implicit val system = mat.system
        implicit val ec: ExecutionContext = system.dispatcher
        implicit val timeout: Timeout = Timeout(5.seconds) // 调整超时时间
        
        val responsePromise = Promise[HttpResponse]()
        
        // 创建临时Actor，用来接收流返回的响应
        val replyActor = system.actorOf(Props(new Actor {
          override def receive: Receive = {
            case response: HttpResponse =>
              responsePromise.success(response)
              context.stop(self) // 完成后停止临时Actor
            case ex: Throwable =>
              responsePromise.failure(ex)
              context.stop(self)
          }
        }))
        
        // 发送请求到共享流的源Actor
        streamSourceActor ! ProcessRequest(req, replyActor)
        
        // 返回响应Future
        complete(responsePromise.future)
      }
    }
  }
}

关键注意事项 #

1. 错误处理：在流的处理逻辑中添加recover或recoverWith，确保处理失败时能正确将错误返回给客户端，避免Promise一直处于pending状态。
2. 流的重启：如果流可能因为异常终止，可以用RestartSource包装源，确保流自动重启，比如：

   import akka.stream.scaladsl.RestartSource
   
   val (requestQueue, streamCompletion) = RestartSource.withBackoff(
     minBackoff = 1.second,
     maxBackoff = 30.seconds,
     randomFactor = 0.2
   )(() => Source.queue[QueuedRequest](100, OverflowStrategy.dropNew))
   .via(processingFlow)
   .to(Sink.ignore)
   .run()
   

3. 缓冲区和并发：根据你的业务并发量调整bufferSize和overflowStrategy，避免请求被过度丢弃或积压。

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