java - 使用 Java Spring Boot 框架订阅 STOMP 消息

Question

我正在尝试编写一个 Spring 服务，该服务订阅外部只读 STOMP 代理并读取/处理它发布的消息。

消息由铁路公司推送到主题“/topic/TRAIN_MVT_ALL_TOC”。我可以成功连接到该主题，但似乎无法实例化其消息的监听器。

我已经设置了一个 Spring @Configuration 类来连接到此，并且在运行应用程序后，它似乎连接正确。

我还创建了消息处理例程，使用 @MessageMapping 注释来监听我感兴趣的特定主题(“TRAIN_MVT_ALL_TOC”)。问题是它似乎永远不会被调用。

配置类代码:`

@Configuration
@EnableWebSocketMessageBroker
public class StompConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/TRAIN_MVT_ALL_TOC").withSockJS();
    }
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.setApplicationDestinationPrefixes("/app");
        registry.enableStompBrokerRelay("/topic")
                .setRelayHost("datafeeds.networkrail.co.uk")
                .setRelayPort(61618)
                .setSystemLogin("MYEMAIL")
                .setSystemPasscode("MYPASSWORD")
        ;
    }
}

消息处理程序代码:

@MessageMapping("/TRAIN_MVT_ALL_TOC")
public void onMessage(@Payload String message) throws Exception {
    System.out.println(message);
}

控制台输出如下日志，表示连接成功。

o.s.w.s.c.WebSocketMessageBrokerStats : WebSocketSession[0 current WS(0)-HttpStream(0)-HttpPoll(0), 0 total, 0 closed abnormally (0 connect failure, 0 send limit, 0 transport error)], stompSubProtocol[processed CONNECT(0)-CONNECTED(0)-DISCONNECT(0)], stompBrokerRelay[1 sessions, ReactorNettyTcpClient[TcpClient: connecting to datafeeds.networkrail.co.uk:61618] (available), processed CONNECT(1)-CONNECTED(1)-DISCONNECT(0)], inboundChannel[pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0], outboundChannelpool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0], sockJsScheduler[pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]

但是该消息永远不会被打印。几天来我一直在努力弄清楚这个问题的真相，因此我们将非常感谢任何帮助。

Answer 1

The messages are pushed to the topic "/topic/TRAIN_MVT_ALL_TOC" by a rail company. I can successfully connect to the topic, but can't seem to be able to instantiate a listener to its messages.

你的意思是客户端的监听器，例如sockjs客户端？

@MessageMapping("/TRAIN_MVT_ALL_TOC")
public void onMessage(@Payload String message) throws Exception {
    System.out.println(message);
}

您不返回任何内容，您需要将其发送到这样的主题:

@MessageMapping("/TRAIN_MVT_ALL_TOC")
@SendTo("/topic/TRAIN_MVT_ALL_TOC")
public Greeting onMessage(HelloMessage message) throws Exception {
    return new Greeting("hello");
}

或者，如果你的构造函数的参数中有一个 SimpMessageSendingOperations (应该由 spring boot 本身自动连接)，你可以向同一主题发送多个消息，如下所示:

@Autowired
public Constructor(SimpMessageSendingOperations messagingTemplate) {
    this.messagingTemplate = messagingTemplate;
}

@MessageMapping(WebSockets.READER_MAPPING)
public void streamOverWebsocket(HelloMessage message) throws Throwable {
    String topicUrl = "/topic/TRAIN_MVT_ALL_TOC";
    messagingTemplate.convertAndSend(topicUrl, new Message("response 1"));
    messagingTemplate.convertAndSend(topicUrl, new Message("response 2"));
    ...
}

最好将传入和传出的内容包装在定义的类中。这样序列化和反序列化就更容易。

来源:

• https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/messaging/simp/SimpMessageSendingOperations.html
• https://spring.io/guides/gs/messaging-stomp-websocket/