使用 Spring 踩踏 websocket 和 sockJS 消息丢失答案

【问题标题】：Stomp over websocket using Spring and sockJS message lost使用 Spring 踩踏 websocket 和 sockJS 消息丢失
【发布时间】：2015-05-25 12:21:51
【问题描述】：

在客户端 javascript 我有

    stomp.subscribe("/topic/path", function (message) {
        console.info("message received");
    });

在服务器端

public class Controller {
  private final MessageSendingOperations<String> messagingTemplate;
  ＠Autowired
  public Controller(MessageSendingOperations<String> messagingTemplate) {
      this.messagingTemplate = messagingTemplate;
  }
  @SubscribeMapping("/topic/path")
  public void subscribe() {
     LOGGER.info("before send");
     messagingTemplate.convertAndSend(/topic/path, "msg");
  }
}

在此设置中，我偶尔（大约在 30 次页面刷新中出现一次）遇到消息丢失，这意味着我在客户端既看不到“收到消息”消息，也看不到来自 Chrome 调试工具的 websocket 流量。

“发送前”始终记录在服务器端。

当我在 subscribe() 方法中调用 MessageSendingOperations 时，它看起来还没有准备好。（如果我把 Thread.sleep(50); 在调用 messingTemplate.convertAndSend 之前，问题就会消失（或者更不可能被重现））

我想知道以前是否有人遇到过同样的情况，是否有事件可以告诉我 MessageSendingOperations 是否准备好。

【问题讨论】：

dom准备好后是否执行stomp.subscribe？
@ᴳᵁᴵᴰᴼ 是的。这是正确的。我可以看到订阅消息是从 Chrome 调试 websocket 网络流量发送的。所以我不认为这是客户端的问题。

标签： spring websocket stomp sockjs spring-websocket

【解决方案1】：

您面临的问题在于clientInboundChannel 的性质，默认情况下为ExecutorSubscribableChannel。

它有3个subscribers：

0 = {SimpleBrokerMessageHandler@5276} "SimpleBroker[DefaultSubscriptionRegistry[cache[0 destination(s)], registry[0 sessions]]]"
1 = {UserDestinationMessageHandler@5277} "UserDestinationMessageHandler[DefaultUserDestinationResolver[prefix=/user/]]"
2 = {SimpAnnotationMethodMessageHandler@5278} "SimpAnnotationMethodMessageHandler[prefixes=[/app/]]"

在taskExecutor 中调用，因此是异步的。

这里的第一个（SimpleBrokerMessageHandler（或StompBrokerRelayMessageHandler，如果你使用broker-relay）负责为topic注册subscription。

您的messagingTemplate.convertAndSend(/topic/path, "msg") 操作可能在该 WebSocket 会话的订阅注册之前执行，因为它们是在单独的线程中执行的。因此，代理处理程序不知道您将消息发送到会话。

@SubscribeMapping 可以在带有return 的方法上配置，该方法的结果将作为回复发送给客户端上的subscription 函数。

HTH

【讨论】：

在我的 subscribe 方法中，我异步调用了服务层，例如 subscriableService.registerAndHandleWith(new Handler(){})。这样我就不能立即用这种方法返回。在这种情况下你会推荐什么？谢谢。
Future.get() 或 CountDownLatch 来自 registerAndHandleWith。从另一边，您可以访问SimpleBrokerMessageHandler 并填充DefaultSubscriptionRegistry 的一些自定义实现以覆盖addSubscriptionInternal 以引发一些自定义ApplicationEvent 以从另一个组件收听它以将该消息发送到主题，当subscription 已经在那里了。以防万一，当您确实需要异步内容并且不要重载 clientInboundChannel 执行程序以等待 Future 时。
再次感谢 Artem，我假设我不能使用现有的 SessionSubscribeEvent 因为它只告诉我客户端已请求但并不意味着订阅注册已完成（如果有像 SessionSubscribedEvent)。
是的，你是对的：在真正的订阅完成之前会发出 SessionSubscribeEvent 事件。随时为SessionSubscribedEvent 提出JIRA 票。

【解决方案2】：

有点晚了，但我想我会添加我的解决方案。在通过消息传递模板发送数据之前，我遇到了订阅未注册的相同问题。由于与DefaultSubscriptionRegistry 的竞争，此问题很少发生且无法预测。

不幸的是，我不能只使用@SubscriptionMapping 的返回方法，因为我们使用的是根据用户类型动态更改的自定义对象映射器（本质上是属性过滤）。

我搜索了 Spring 代码，发现 SubscriptionMethodReturnValueHandler 负责发送订阅映射的返回值，并且与我的异步控制器的自动连接的 SimpMessagingTemplate 具有不同的消息传递模板！！

所以解决方案是将MessageChannel clientOutboundChannel 自动连接到我的异步控制器中，并使用它来创建SimpMessagingTemplate。（您不能直接将其连接起来，因为您只会将模板发送给代理）。

在订阅方法中，我使用了直接模板，而在其他方法中，我使用了发送给代理的模板。

【讨论】：

【解决方案3】：

这是我的解决方案。它是沿着相同的路线。添加了ExecutorChannelInterceptor 并发布了自定义SubscriptionSubscribedEvent。关键是在消息被 AbstractBrokerMessageHandler 处理后发布事件，这意味着订阅已经向代理注册。

@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
    registration.interceptors(new ExecutorChannelInterceptorAdapter() {

        @Override
        public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
            SimpMessageHeaderAccessor accessor = SimpMessageHeaderAccessor.wrap(message);
            if (accessor.getMessageType() == SimpMessageType.SUBSCRIBE && handler instanceof AbstractBrokerMessageHandler) {
                /*
                 * Publish a new session subscribed event AFTER the client
                 * has been subscribed to the broker. Before spring was
                 * publishing the event after receiving the message but not
                 * necessarily after the subscription occurred. There was a
                 * race condition because the subscription was being done on
                 * a separate thread.
                 */
                applicationEventPublisher.publishEvent(new SessionSubscribedEvent(this, message));
            }
        }
    });

}

【讨论】：