User Destinations / 用户目的地

User Destinations

应用程序可以发送针对特定用户的消息，Spring 的 STOMP 支持为此目的识别以 /user/为前缀的目标。例如，一个客户端可能会订阅 /user/queue/position-updates目的地。UserDestinationMessageHandler 处理这个目的地，并将其转换为用户会话的唯一目的地（如 /queue/position-updates-user123）。这为订阅一个通用命名的目的地提供了便利，同时，确保不会与订阅同一目的地的其他用户发生冲突，以便每个用户都能收到独特的股票位置更新。

:::warning 当使用用户目的地时，重要的是配置消息代理（broker）和应用程序的目的地前缀，如 启用 STOMP 中所示，否则消息代理将处理 /user前缀的消息，而这些消息只应由 UserDestinationMessageHandler 处理。 :::

在发送方面，消息可以被发送到一个目的地，如 /user/{username}/queue/position-updates，这又被UserDestinationMessageHandler 翻译成一个或多个目的地，一个用于与用户相关的会话。这让应用程序中的任何组件都可以发送针对特定用户的消息，而不一定要知道他们的名字和通用目的地。这也是通过一个注解和一个消息模板来支持的。

一个消息处理方法可以通过 @SendToUser注解（也支持在类级别上共享一个共同的目的地）向与被处理的消息相关的用户发送消息，如下例所示：

@Controller
public class PortfolioController {
    @MessageMapping("/trade")
    @SendToUser("/queue/position-updates")
    public TradeResult executeTrade(Trade trade, Principal principal) {
        // ...
        return tradeResult;
    }
}

如果用户有一个以上的会话，默认情况下，所有订阅给定目的地的会话都是目标。然而，有时可能需要只针对发送被处理消息的会话。你可以通过将广播属性设置为 false 来做到这一点，正如下面的例子所示：

@Controller
public class MyController {
    @MessageMapping("/action")
    public void handleAction() throws Exception{
        // 在此引发 MyBusinessException
    }
    @MessageExceptionHandler
    @SendToUser(destinations="/queue/errors", broadcast=false)
    public ApplicationError handleException(MyBusinessException exception) {
        // ...
        return appError;
    }
}

:::info 虽然用户目的地通常意味着有一个 经过认证的用户，但这并不是严格的要求。未与认证用户相关联的 WebSocket 会话可以订阅用户目的地。在这种情况下，@SendToUser注解的行为与 broadcast=false 时完全相同（即只针对发送被处理消息的会话）。 :::

你可以从任何应用组件向用户目的地发送消息，例如，通过注入由 Java 配置或 XML 命名空间创建的 SimpMessagingTemplate。(如果需要用 @Qualifier来限定，bean 的名字是 brokerMessagingTemplate)。下面的例子显示了如何做到这一点：

@Service
public class TradeServiceImpl implements TradeService {
    private final SimpMessagingTemplate messagingTemplate;
    @Autowired
    public TradeServiceImpl(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }
    // ...
    public void afterTradeExecuted(Trade trade) {
        this.messagingTemplate.convertAndSendToUser(
                trade.getUserName(), "/queue/position-updates", trade.getResult());
    }
}

:::info 当您将 用户目的地 与 外部消息代理 一起使用时，您应该查看关于如何管理非活动队列的代理文档，以便在用户会话结束时，所有独特的用户队列都被删除。例如，当您使用诸如 /exchange/amq.direct/position-updates 等目的地时，RabbitMQ 会创建自动删除队列。因此，在这种情况下，客户端可以订阅到 /user/exchange/amq.direct/position-updates。同样地，ActiveMQ 也有清除非活动目的地的 配置选项。 :::

在一个多应用服务器的情况下，一个用户目的地可能会因为用户连接到一个不同的服务器而保持未解决。在这种情况下，你可以配置一个目的地来广播未解决的消息，以便其他服务器有机会尝试。这可以通过 Java 配置中 MessageBrokerRegistry 的userDestinationBroadcast 属性和 XML 中 message-broker 元素的 user-destination-broadcast 属性完成。

一个例子

在 前面的例子 基础上进行

用户队列持久化方式

在配置中开启 /user 用户目的地处理配置。

package cn.mrcode.study.springdocsread.websocket;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy;
import java.security.Principal;
/**
 * @author mrcode
 */
@Configuration
@EnableWebSocketMessageBroker
public class MyWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // portfolio 是 WebSocket（或 SockJS）的端点的 HTTP URL。客户端需要连接以进行 WebSocket 握手。
        registry.addEndpoint("/portfolio")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // 目标标头以 /app 开头的 STOMP 消息会被路由到 @Controller 类中的 @MessageMapping 方法。
        config.setApplicationDestinationPrefixes("/app");
        /*
            启用用户目的地，它的原理如下（这个在源码的注释上有说明）：
            配置用于标识用户目的地的前缀。用户目的地为用户提供了订阅其会话唯一的队列名称以及其他人将消息发送到那些唯一的、特定于用户的队列的能力。
            例如，当用户尝试订阅“/user/queue/position-updates”时，目的地可能会被转换为“/queue/position-updatesi9oqdfzo”，从而产生一个唯一的队列名称，不会与任何其他尝试订阅的用户发生冲突相同。随后，当消息发送到“/user/{username}/queue/position-updates”时，目的地被转换为“/queue/position-updatesi9oqdfzo”。
            用于标识此类目的地的默认前缀是“/user/”。
         */
        config.setUserDestinationPrefix("/user");
        // 使用内置的消息代理进行订阅和广播，并且 将目标标头以 /topic 或 /queue 开头的消息路由到代理。
        config.enableStompBrokerRelay("/topic", "/queue")
                // 默认使用 ReactorNettyTcpClient
                // .setTcpClient()
                // 所以可以直接设置相关的属性
                .setRelayHost("127.0.0.1")
                .setRelayPort(61613)
        ;
    }
    /**
     * 入栈 通道配置
     *
     * @param registration
     */
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        // 添加拦截器
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                // 从消息中获取与 stomp 相关的请求头
                StompHeaderAccessor accessor =
                        MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                System.out.println(accessor.getCommand());
                // 如果当前的消息命令类型是 链接 类型，则处理 认证相关的信息
                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    // 获取登录用户 和 密码 header
                    final String login = accessor.getLogin();
                    final String passcode = accessor.getPasscode();
                    // 如果没有设置 login 头就抛出异常
                    if (!StringUtils.hasLength(login)) {
                        throw new RuntimeException("必须设置用户名");
                    }
                    // 在这里你可以查询数据库之类的方式验证
                    // 注意：这个 Authentication 是 security 里面的类
                    // Authentication user = ... ; // access authentication header(s)
                    // 但是这里要求的是 java.security.Principal 就可以，所以我们可以自己实现一个
                    accessor.setUser(new Principal() {
                        @Override
                        public String getName() {
                            return login;
                        }
                    });
                }
                return message;
            }
        });
    }
    /**
     * spring 为了支持每种容器自己的 websocket 升级策略，抽象了 RequestUpgradeStrategy，
     * <p>对 tomcat 提供了 TomcatRequestUpgradeStrategy 策略</p>
     * 如果不申明这个，就会在启动的时候抛出异常：No suitable default RequestUpgradeStrategy found
     */
    @Bean
    public TomcatRequestUpgradeStrategy tomcatRequestUpgradeStrategy() {
        return new TomcatRequestUpgradeStrategy();
    }
}

然后编写一个接受请求更新，并将结果转发给发起请求用户的 @MessageMapping 方法

package cn.mrcode.study.springdocsread.websocket;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.simp.annotation.SendToUser;
import org.springframework.messaging.simp.annotation.SubscribeMapping;
import org.springframework.stereotype.Controller;
import java.security.Principal;
/**
 * @author mrcode
 */
@Controller
public class StompController {
    /**
     * @param greeting
     * @return 返回值是广播给所有人
     */
    // 需要注意的是：客户端需要发送消息到 /app/greeting
    // 响应的消息，会默认广播到  /topic/greeting  上，只要订阅了 /topic/greeting 的订阅者都能收到
    @MessageMapping("/greeting")
    public String handle(String greeting) {
        return "[" + getTimestamp() + ": " + greeting;
    }
    private String getTimestamp() {
        return System.currentTimeMillis() + "";
    }
    /**
     * @return 返回值只返回给订阅的人;
     */
    // 需要注意的是：前端需要订阅  /app/greeting2
    // 也就是说，只要有订阅 /app/greeting2，订阅成功后，该订阅者就会收到这里返回的消息
    @SubscribeMapping("/greeting2")
    public String handle2() {
        return "[ 单个消息" + getTimestamp() + ": ";
    }
    // 每个用户执行更新操作，然后将更新的结果反馈给该用户
    // 首先客户端需要订阅 /user/queue/trade
    // 客户端发送消息到到 /app/trade 更新消息
    // 服务器随后会将更新后的结果推送给 /user/{username}/queue/trade , 该用户就能获取到消息了
    @MessageMapping("/trade")
    @SendToUser("/queue/trade")
    public String executeTrade(String mesg, Principal principal) {
        // ...
        return "消息已经处理完成：" + mesg;
    }
}

前端页面订阅和发起消息

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>STOMP</title>
</head>
<body>
</body>
<script type="text/javascript" src="/node_modules/webstomp-client/dist/webstomp.min.js"></script>
<script type="text/javascript"
        src="https://cdnjs.loli.net/ajax/libs/sockjs-client/1.6.0/sockjs.js"></script>
<script>
    // 这里使用 sockJs 库链接
    var socket = new SockJS("http://localhost:8080/portfolio");
    // 文章上说可以使用 WebSocket 链接。 实际上我这里测试不可以，会报错
    // var socket = new WebSocket("ws://localhost:8080/portfolio");
    var stompClient = webstomp.over(socket);
    // 链接上 服务器时
    stompClient.connect(
        // 头信息
        {
            login: 'user1',
            passcode: '123456',
        },
        // 链接成功回调函数
        function (frame) {
            console.log(frame)
            // // 订阅消息
            // stompClient.subscribe("/topic/greeting", msg => {
            //     console.log("收到订阅的消息广播：" + msg.body)
            // })
            //
            // // 订阅消息
            // stompClient.subscribe("/app/greeting2", msg => {
            //     console.log("收到初始化的订阅消息：" + msg.body)
            // })
            //
            // // 链接上服务器时,像服务器发送一个消息
            // stompClient.send("/app/greeting", "我的第一个消息")
            // 订阅消息
            // stompClient.subscribe("/user/exchange/amq.direct/position-updates", msg => {
            //     console.log("收到订阅的 /user/queue/trade 消息：" + msg.body)
            // })
            // 这里为了更好的看到控制台的信息，将上面的测试都注释掉了
            stompClient.subscribe("/user/queue/trade", msg => {
                console.log("收到订阅的 /user/queue/trade 消息：" + msg.body)
            })
            // 发送更新消息
            stompClient.send("/app/trade", "更新消息")
        },
        // 链接失败函数
        function (frame) {
            console.log("链接失败：", frame)
            // 如果是一个 stomp 帧，这判定 command ，如果没有这可能是一个 CloseEvent 对象
            if (frame.command && frame.command == 'ERROR') {
                console.log("链接失败原因：", frame.headers.message)
            }
        }
    )
</script>
</html>

然后启动测试，打开该 html 页面，控制台可以看到有如下的输出

如果你将该 html 复制一份，并将 login 修改成 user2 打开后，就能看到只有当前的用户能接受到这个对应的订阅消息，其他用户是看不到的。

然后再来看看 RabbitMQ 里面发生了什么

和本章文档说的一样，生成了一个单独的队列，不过它不是按照用户名生成的，而是按照 simpSessionId（可以在后端的用户登录认证地方，也就是 StompHeaderAccessor 中的 headers 中看到这个值）

所以这里我就没有明白为什么是 session 还需要搞一个持久化的队列呢？难道可以自定义 sessionId 的生成吗？不然这个持久化的队列貌似没有什么作用，每次刷新页面生成的 sessionId 都是不一样的

用户临时队列

前面看到了，使用 /user/queue 方式会生成一个持久化的队列。目前没有搞明白 sessionId 的情况下，持久化队列不会自动删除，刷新一次页面就会生成一次，这就严重的不太方便。

可以使用 /exchange方式生成临时队列，先来看如何配置，再讲原理

package cn.mrcode.study.springdocsread.websocket;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompCommand;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy;
import java.security.Principal;
/**
 * @author mrcode
 */
@Configuration
@EnableWebSocketMessageBroker
public class MyWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // portfolio 是 WebSocket（或 SockJS）的端点的 HTTP URL。客户端需要连接以进行 WebSocket 握手。
        registry.addEndpoint("/portfolio")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        // 目标标头以 /app 开头的 STOMP 消息会被路由到 @Controller 类中的 @MessageMapping 方法。
        config.setApplicationDestinationPrefixes("/app");
        /*
            启用用户目的地，它的原理如下（这个在源码的注释上有说明）：
            配置用于标识用户目的地的前缀。用户目的地为用户提供了订阅其会话唯一的队列名称以及其他人将消息发送到那些唯一的、特定于用户的队列的能力。
            例如，当用户尝试订阅“/user/queue/position-updates”时，目的地可能会被转换为“/queue/position-updatesi9oqdfzo”，从而产生一个唯一的队列名称，不会与任何其他尝试订阅的用户发生冲突相同。随后，当消息发送到“/user/{username}/queue/position-updates”时，目的地被转换为“/queue/position-updatesi9oqdfzo”。
            用于标识此类目的地的默认前缀是“/user/”。
         */
        config.setUserDestinationPrefix("/user");
        // 使用内置的消息代理进行订阅和广播，并且 将目标标头以 /topic 或 /queue 开头的消息路由到代理。
        // 这里增加 /exchange 的前缀
        config.enableStompBrokerRelay("/topic", "/queue", "/exchange")
                // 默认使用 ReactorNettyTcpClient
                // .setTcpClient()
                // 所以可以直接设置相关的属性
                .setRelayHost("127.0.0.1")
                .setRelayPort(61613)
        ;
    }
    /**
     * 入栈 通道配置
     *
     * @param registration
     */
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        // 添加拦截器
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                // 从消息中获取与 stomp 相关的请求头
                StompHeaderAccessor accessor =
                        MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                System.out.println(accessor.getCommand());
                // 如果当前的消息命令类型是 链接 类型，则处理 认证相关的信息
                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    // 获取登录用户 和 密码 header
                    final String login = accessor.getLogin();
                    final String passcode = accessor.getPasscode();
                    // 如果没有设置 login 头就抛出异常
                    if (!StringUtils.hasLength(login)) {
                        throw new RuntimeException("必须设置用户名");
                    }
                    // 在这里你可以查询数据库之类的方式验证
                    // 注意：这个 Authentication 是 security 里面的类
                    // Authentication user = ... ; // access authentication header(s)
                    // 但是这里要求的是 java.security.Principal 就可以，所以我们可以自己实现一个
                    accessor.setUser(new Principal() {
                        @Override
                        public String getName() {
                            return login;
                        }
                    });
                }
                return message;
            }
        });
    }
    /**
     * spring 为了支持每种容器自己的 websocket 升级策略，抽象了 RequestUpgradeStrategy，
     * <p>对 tomcat 提供了 TomcatRequestUpgradeStrategy 策略</p>
     * 如果不申明这个，就会在启动的时候抛出异常：No suitable default RequestUpgradeStrategy found
     */
    @Bean
    public TomcatRequestUpgradeStrategy tomcatRequestUpgradeStrategy() {
        return new TomcatRequestUpgradeStrategy();
    }
}

前端页面

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>STOMP</title>
</head>
<body>
</body>
<script type="text/javascript" src="/node_modules/webstomp-client/dist/webstomp.min.js"></script>
<script type="text/javascript"
        src="https://cdnjs.loli.net/ajax/libs/sockjs-client/1.6.0/sockjs.js"></script>
<script>
    // 这里使用 sockJs 库链接
    var socket = new SockJS("http://localhost:8080/portfolio");
    // 文章上说可以使用 WebSocket 链接。 实际上我这里测试不可以，会报错
    // var socket = new WebSocket("ws://localhost:8080/portfolio");
    var stompClient = webstomp.over(socket);
    // 链接上 服务器时
    stompClient.connect(
        // 头信息
        {
            login: 'user1',
            passcode: '123456',
        },
        // 链接成功回调函数
        function (frame) {
            console.log(frame)
            // // 订阅消息
            // stompClient.subscribe("/topic/greeting", msg => {
            //     console.log("收到订阅的消息广播：" + msg.body)
            // })
            //
            // // 订阅消息
            // stompClient.subscribe("/app/greeting2", msg => {
            //     console.log("收到初始化的订阅消息：" + msg.body)
            // })
            //
            // // 链接上服务器时,像服务器发送一个消息
            // stompClient.send("/app/greeting", "我的第一个消息")
            // 订阅消息
            // stompClient.subscribe("/user/exchange/amq.direct/position-updates", msg => {
            //     console.log("收到订阅的 /user/queue/trade 消息：" + msg.body)
            // })
            // stompClient.subscribe("/user/queue/trade", msg => {
            //     console.log("收到订阅的 /user/queue/trade 消息：" + msg.body)
            // })
            // // 发送更新消息
            // stompClient.send("/app/trade", "更新消息")
            stompClient.subscribe("/user/exchange/amq.direct/trade", msg => {
                console.log("收到订阅的 /user/exchange/amq.direct/trade 消息：" + msg.body)
            })
            // 这里不延迟一会发送消息，可能会导致订阅的临时队列还没有创建完成，从而表现出来是发送了消息，但是没有接受到消息的情况
            setTimeout(()=>{
                // 发送更新消息
                stompClient.send("/app/trade", "更新消息")
            },500)
        },
        // 链接失败函数
        function (frame) {
            console.log("链接失败：", frame)
            // 如果是一个 stomp 帧，这判定 command ，如果没有这可能是一个 CloseEvent 对象
            if (frame.command && frame.command == 'ERROR') {
                console.log("链接失败原因：", frame.headers.message)
            }
        }
    )
</script>
</html>

看看测试的 html 控制台

然后再看看 RabbitMQ 中的信息

与前面的持久化队列方式相差太多了。 这里队列名称里面都没有 trade 了。而且是临时队列了。

解密用户临时队列的原理

在 RabbitMQ STOMP 官方文档 中有如下的描述

可以看到，目的地前缀支持上面的好多种，所以我们需要在 enableStompBrokerRelay 中开启对这个前缀的支持，否则就不会转发到对应的消息代理上

        config.enableStompBrokerRelay("/topic", "/queue", "/exchange")

然后再来看 exchange/amq.direct/trade这个地址的含义， AMQP 的语法如下

/exchange/exchange_name [/routing_key]

所以对于我们的地址来说就很明显了：

• /exchange 消息代理固定前缀
• amq.direct这个是 RabbitMQ 的默认 Exchange ，可以在 exchanges 页面看到，如下图所示

• trad ：就是 routing_key 了

明白了路径的含义之后，你一定在想，这也看不出来和是如何与一个用户的会话 ID 对应起来的。这个只需要点开这个临时队列就清楚了

在这个临时队列绑定到 amq.direct exchange 的时候，将 routing_key 设置成了之前持久化队列的名称。 这下你就真的明白了他们的关系是如何对应的了