为什么Akka HTTP会在生成多条消息时关闭用户连接？

Posted 2021-04-12

我有一个简单的WebSocket应用程序，它基于Akka HTTP / Reactive streams，就像这个https://github.com/calvinlfer/akka-http-streaming-response-examples/blob/master/src/main/scala/com/experiments/calvin/ws/WebSocketRoutes.scala#L82。

换句话说，我有Sink，Source（由Publisher制作）和Flow：

Flow.fromSinkAndSource(incomingMessages, outgoingMessages)

当我每秒向客户端生成超过30条消息时，Akka关闭了一个连接。

我无法理解，配置此行为的设置在哪里。我知道OverflowStrategy，但我没有明确配置它。

看来，我有OverflowStrategy.fail()，或者我的问题看起来像它。

答案

你可以调整Internal buffers。

有两种方法，如何做到：

1）application.conf：

akka.stream.materializer.max-input-buffer-size = 1024

2）您可以为您的Flow明确配置它：

Flow.fromSinkAndSource(incomingMessages, outgoingMessages)
  .addAttributes(Attributes.inputBuffer(initial = 1, max = 1024))