多服务器环境中的用户目标？ （Spring WebSocket和RabbitMQ）

Posted 2021-04-16

Spring WebSockets的documentation声明：

4.4.13。用户目的地

应用程序可以发送针对特定用户的消息，Spring的STOMP支持可以识别以“/ user /”为前缀的目标。例如，客户端可能订阅目标“/ user / queue / position-updates”。该目的地将由UserDestinationMessageHandler处理并转换为用户会话唯一的目的地，例如， “/队列/位置更新-user123”。这提供了订阅一般命名的目的地的便利性，同时确保不与订阅相同目的地的其他用户发生冲突，使得每个用户可以接收唯一的库存位置更新。

这应该在RabbitMQ作为代理的多服务器环境中工作吗？

据我所知，用户的队列名称是通过附加simpSessionId生成的。当使用推荐的客户端库stomp.js时，这会导致第一个用户获取队列名称"/queue/position-updates-user0"，下一个获取"/queue/position-updates-user1"，依此类推。这反过来意味着连接到不同服务器的第一个用户将订阅相同的队列（"/queue/position-updates-user0"）。

我在文档中可以找到的唯一参考是：

在多应用程序服务器方案中，用户目标可能仍未解析，因为用户连接到不同的服务器。在这种情况下，您可以配置目标以广播未解析的消息，以便其他服务器有机会尝试。这可以通过Java config中的MessageBrokerRegistry的userDestinationBroadcast属性和XML中的message-broker元素的user-destination-broadcast属性来完成。

但这只能使得与来自与建立Web套接字的服务器不同的服务器的用户进行通信成为可能。

我觉得我错过了什么？反正配置Spring是否能够在多服务器环境中安全地使用MessagingTemplate.convertAndSendToUser(principal.getName(), destination, payload)？

答案

如果需要进行身份验证（我假设他们的凭据存储在数据库中），您始终可以使用他们的数据库唯一用户ID进行订阅。

我所做的是当用户登录时，他们会自动订阅两个主题，一个用于系统范围广播的account|system主题和用于特定广播的account|<userId>主题。

您可以尝试像notification|<userid>这样的人为每个人订阅然后发送消息到该主题，他们将收到它。

由于用户ID对每个用户都是唯一的，因此只要每个环境都访问相同的数据库信息，您就不应该在群集环境中出现问题。

这是我的发送方法：

  public static boolean send(Object msg, String topic) {
    try {
      String destination = topic;
      String payload = toJson(msg); //jsonfiy the message 
      Message<byte[]> message = MessageBuilder.withPayload(payload.getBytes("UTF-8")).build();
      template.send(destination, message);
      return true;
    } catch (Exception ex) {
      logger.error(CommService.class.getName(), ex);
      return false;
    }
  }

我的目的地已预先格式化，因此，如果我想向id为1的用户发送消息，目的地看起来像/topic/account|1。

我创建了一个乒乓控制器，为连接的用户测试websockets，看看他们的环境是否允许websockets。我不知道这是否会对您有所帮助，但这在我的集群环境中是有效的。

/**
   * Play ping pong between the client and server to see if web sockets work
   * @param input the ping pong input
   * @return the return data to check for connectivity
   * @throws Exception exception
   */
  @MessageMapping("/ping")
  @SendToUser(value="/queue/pong", broadcast=false) // send only to the session that sent the request
  public PingPong ping(PingPong input) throws Exception {
    int receivedBytes = input.getData().length;
    int pullBytes = input.getPull();

    PingPong response = input;
    if (pullBytes == 0) {
      response.setData(new byte[0]);
    } else if (pullBytes != receivedBytes)  {
      // create random byte array
      byte[] data =  randomService.nextBytes(pullBytes);
      response.setData(data);
    }
    return response;
  }