IllegalStateException - Sink.asPublisher 仅支持一个订阅 - 使用 WebSockets 时

问题描述 投票:0回答:1

在生产中,我们的应用程序生成了以下堆栈跟踪。这里没有我们自己的代码 - 没有太多可继续的。

了解

akka-stream
库的人可以解释一下吗:

  1. 这里发生了什么事?
  2. 这只是 Play 中的一个潜在错误吗?我们尚未在相当相似的测试环境中重现它,因此按照 PlayFramework GitHub 的要求生成测试用例是不切实际的。

播放版本:2.8.18 阿卡版本:2.6.20

2023-11-06 09:20:05 GMT [ERROR] p.c.s.c.WebSocketFlowHandler [WebSocketFlowHandler.scala:249] - WebSocket flow threw exception
java.lang.IllegalStateException: Sink.asPublisher(fanout = false) only supports one subscriber (which is allowed, see reactive-streams specification, rule 1.11)
        at akka.stream.impl.ReactiveStreamsCompliance$.rejectAdditionalSubscriber(ReactiveStreamsCompliance.scala:62)
        at akka.stream.impl.VirtualPublisher.rec$6(StreamLayout.scala:481)
        at akka.stream.impl.VirtualPublisher.subscribe(StreamLayout.scala:486)
        at akka.stream.impl.fusing.ActorGraphInterpreter$BatchingActorInputBoundary.preStart(ActorGraphInterpreter.scala:148)
        at akka.stream.impl.fusing.GraphInterpreter.init(GraphInterpreter.scala:306)
        at akka.stream.impl.fusing.GraphInterpreterShell.init(ActorGraphInterpreter.scala:619)
        at akka.stream.impl.fusing.ActorGraphInterpreter.tryInit(ActorGraphInterpreter.scala:727)
        at akka.stream.impl.fusing.ActorGraphInterpreter.finishShellRegistration(ActorGraphInterpreter.scala:770)
        at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$shortCircuitBatch(ActorGraphInterpreter.scala:788)
        at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:819)
playframework akka-stream
1个回答
0
投票

如果没有看到你的代码(即你想要实现的目标),就很难知道出了什么问题。但它表明,在某个地方,您已经获取了由

Publisher
生成的
Sink.asPublisher
,并尝试订阅它两次。这是不可能的,因为
Sink.asPublisher
返回的发布者代表流的一个正在运行的实例,并且如果您第二次订阅,则无法为第二个订阅者重新启动该流。

从概念上思考,当您收到 WebSocket 连接,并且想要使用传入的消息流时,您可以提供一个接收器来执行此操作。

Sink.asPublisher
将该流转换为反应式流发布者,您可以通过其他反应式流库使用来自它的消息。当您订阅它时,提供的订阅者将连接到现有的 WebSocket 连接。但是,如果您再次尝试从该发布者那里订阅,它会做什么呢? WebSocket 服务器无法返回到 WebSocket 客户端并说“嘿,你可以再次连接到我,以便我可以将另一个消息流发送给这个新订阅者吗?”不行,只能失败,说不能订阅两次。

© www.soinside.com 2019 - 2024. All rights reserved.