java - 如何将convertAndSendToUser()与外部代理(例如Spring4中的RabbitMQ)一起使用？

Question

我尝试使用 RabbitMQ 作为外部代理在 Spring 4 中配置 Web 套接字支持，但是一旦我切换到 RabbitMQ，我就会在客户端启动时收到以下错误:

'/user/queue/changes' is not a valid destination.
Valid destination types are: /temp-queue, /exchange, /topic, /queue, /amq/queue, /reply-queue/.

在服务器上，我使用convertAndSendToUser，这对于简单的代理来说工作得很好，但是一旦我切换到 RabbitMQ，我就会收到此错误。请注意，RabbitMQ 对于普通主题广播工作得很好 - 只是 /user channel 崩溃了。

我需要做一些特殊的事情才能让 /user 与 RabbitMQ 一起使用吗？

编辑以包含 Web Socket 配置

我的 WebSocketConfig 非常标准，需要进行一些自定义才能与 spring-session 集成。 :

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractSessionWebSocketMessageBrokerConfigurer<ExpiringSession> {


@Override
public void configureStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/changes").withSockJS();
}

@Override
public void configureMessageBroker(MessageBrokerRegistry config) {
    //config.enableSimpleBroker("/queue", "/topic");
    StompBrokerRelayRegistration r = config.enableStompBrokerRelay("/user", "/topic");

    try {
        String rabbitUrl = System.getenv("CLOUDAMQP_URL");
        if(rabbitUrl != null) { 
            log.info("RABBIT URL detected: " + rabbitUrl);
            URI uri = new URI(rabbitUrl);
            String host = uri.getHost();                
            String login = uri.getUserInfo().split(":",2)[0];
            String passCode = uri.getUserInfo().split(":",2)[1];
            String vhost = uri.getPath().substring(1);
            r.setRelayHost(host);
            r.setSystemLogin(login);    
            r.setSystemPasscode(passCode);
            r.setClientLogin(login);    
            r.setClientPasscode(passCode);
            r.setVirtualHost(vhost);                
        }
    } catch(Exception e) {
        log.error("Error setting up RabbitMQ", e);
    }
    config.setApplicationDestinationPrefixes("/app");
  }
}

Answer 1

我也遇到了类似的错误，但在网上搜索后并没有找到直观的解决方案。

我想在阅读 RabbitMQ STOMP 的文档后分享我的发现。

根据 RabbitMQ STOMP 的文档，仅以 /exchange 开头的目的地, /queue , /amq/queue , /topic和temp-queue被允许。目的地/user/*不会被允许。您可以根据消息的要求选择目的地。

/exchange -- SEND to arbitrary routing keys and SUBSCRIBE to arbitrary binding patterns;

/queue -- SEND and SUBSCRIBE to queues managed by the STOMP gateway;

/amq/queue -- SEND and SUBSCRIBE to queues created outside the STOMP gateway;

/topic -- SEND and SUBSCRIBE to transient and durable topics;

/temp-queue/ -- create temporary queues (in reply-to headers only).

例如，我想向某个主题发送消息以通知所有订阅者。

For simple topic destinations which deliver a copy of each message to all active subscribers, destinations of the form /topic/< name > can be used. Topic destinations support all the routing patterns of AMQP topic exchanges.

Messages sent to a topic destination that has no active subscribers are simply discarded.

AMQP 0-9-1 Semantics

For SEND frames, the message is sent to the amq.topic exchange with the routing key < name >.

For SUBSCRIBE frames, an autodeleted, non-durable queue is created and bound to the amq.topic exchange with routing key < name >. A subscription is created against the queue.

该规范意味着目标为 /topic/<name> 的 stomp 子消息将使用rabbitmq的默认交换器amp.topic，如果变量名称不存在，则可能会创建一个绑定(bind)，并且还会创建一个队列来绑定(bind)交换器amp.topic 通过该绑定(bind)。

如果您想创建持久订阅，客户端应发送带有以下 header 的订阅消息。

durable:true
auto-delete:false 

在我的应用程序中，websocket 服务器收到消息 /watch/{liveid} ，然后回复主题/topic/watchinfo-{liveid}另一条消息.

@Secured(User.ROLE_USER)
@MessageMapping("/watch/{liveid}")
@SendTo("/topic/watchinfo-{liveid}")
@JsonView(View.Live.class)
public LiveWatchInfoMessage liveinfo(@DestinationVariable("liveid") String liveid,
                                         @AuthenticationPrincipal UserDetails activeUser) {
        ...
        return LiveWatchInfoMessage.builder().build();
    }