使用 scala 和 Akka 在不丢失数据的情况下使用服务器发送事件（SSE）答案

【问题标题】：Consuming Server Sent Events(SSE) without losing data using scala and Akka使用 scala 和 Akka 在不丢失数据的情况下使用服务器发送事件（SSE）
【发布时间】：2021-10-08 05:01:59
【问题描述】：

当生产速率>消费速率时，我想消费 SSE 事件而不丢失任何数据。由于 SSE 支持背压，Akka 应该能够做到。我尝试了几种不同的方法，但多余的消息被丢弃了。


@Singleton
class SseConsumer @Inject()()(implicit ec: ExecutionContext) {

  implicit val system = ActorSystem()

  val send: HttpRequest => Future[HttpResponse] = foo

  def foo(x: HttpRequest) = {
    try {
      val authHeader = Authorization(BasicHttpCredentials("user", "pass"))
      val newHeaders = x.withHeaders(authHeader)
      Http().singleRequest(newHeaders)
    } catch {
      case e: Exception => {
        println("Exceptio12n", e.printStackTrace())
        throw e
      }
    }
  }


  val eventSource2: Source[ServerSentEvent, NotUsed] =
    EventSource(
      uri = Uri("https://xyz/a/events/user"),
      send,
      initialLastEventId = Some("2"),
      retryDelay = 1.second
    )


  def orderStatusEventStable() = {
    val events: Future[immutable.Seq[ServerSentEvent]] =
      eventSource2
        .throttle(elements = 1, per = 3000.milliseconds, maximumBurst = 1, ThrottleMode.Shaping)
        .take(5)
        .runWith(Sink.seq)
    events.map(_.foreach(x => {
      // TODO: push to sqs
      println("456")
      println(x.data)
    }))
  }


  Future {
    blocking {
      while (true) {
        try {
          Await.result(orderStatusEventStable() recover {
            case e: Exception => {
              println("exception", e)
              throw e
            }
          }, Duration.Inf)
        } catch {
          case e: Exception => {
            println("Exception", e.printStackTrace())
          }
        }
      }
    }
  }
}

此代码有效，但存在以下问题：

由于.take(5) 当消费率
另外，我想处理每条消息，并且不想等到 5 条消息到达。我该怎么做？
我必须在一个while循环中编写消费者。这似乎不是基于事件的，而更像是轮询（非常类似于使用分页和限制为 5 调用 GET）
我不确定限制，尝试阅读文档但它非常混乱。如果我不想丢失任何事件，那么节流是正确的方法吗？我预计高峰时段的速率为 5000 req / sec，否则为 10 req / sec。当生产率很高时，理想情况下我想施加背压。节流是正确的方法吗？根据文档，它似乎是正确的，因为它说Backpressures when downstream backpressures or the incoming rate is higher than the speed limit

【问题讨论】：

标签： scala playframework akka reactive-programming akka-stream

【解决方案1】：

为了让 Akka Stream 反压工作，您必须只使用一个源，而不是每次都使用新源重新创建一种轮询。

忘记你的循环和你的def orderStatusEventStable。

只做这样的事情（一次）：

eventSource2
  .operator(event => /* do something */ )
  .otherOperatorMaybe()
  ...
  .runWith(Sink.foreach(println))

operator 和 otherOperatorMaybe 是 Akka Stream 上的操作，具体取决于您想要实现的目标（例如原始代码中的 throttle 和 take）。

运营商列表：https://doc.akka.io/docs/akka/current/stream/operators/index.html

Akka Stream 很强大，但你需要花一些时间来了解它

【讨论】：