从应用程序的其他层发送 STOMP 消息 [英] Sending STOMP messages from other layers of an application

问题描述

我正在使用 RabbitMQ 代理的集群 tomcat 环境中使用 Spring Websockets 构建应用程序.我有一个 API 模块，它需要注册要监听的端点.我按照正常的例子并想出了这个配置:

I am building an application using Spring Websockets on a clustered tomcat environment with a RabbitMQ broker. I have an API module which needs to register the endpoint to listen to. I followed the normal examples and came up with this config:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer
{
    @Override
    public void configureMessageBroker(final MessageBrokerRegistry config)
    {
        config.enableStompBrokerRelay("/topic/")
            .setRelayHost("localhost")
            .setRelayPort(61613)
            .setClientLogin("guest")
            .setClientPasscode("guest");
    }

    @Override
    public void registerStompEndpoints(final StompEndpointRegistry registry)
    {
        registry.addEndpoint("/updates")
            .setAllowedOrigins("*")
            .withSockJS();
    }
}

虽然这有效，但它并没有解决我的问题，因为 WebSocket 和中继配置似乎都捆绑到 API 模块中，因此其他层无法重用代理.我需要在服务层进行 stomp 消息代理中继配置，以便我们应用程序的其他模块可以将消息推送到 RabbitMQ 中的主题，然后转向并通知 API 模块更新所有打开的 websocket.

While this works, it doesn't solve my issue as it appears the WebSocket and relay config are all bundled into the API module therefore leaving other layers unable to reuse the broker. I need the stomp message broker relay configuration to happen at the service layer so that other modules of our app can push messages to topics in RabbitMQ which then turn around and notify the API module to update all open websockets.

下面是我们应用程序中相关层的示例图以及我要完成的任务.我需要允许模块Cron Message Sender"通过我们的其他 API 模块向订阅消息主题的每个人推送消息.

Below is a sample diagram of the relevant layers in our application and what I am trying to accomplish. I need to allow the module "Cron Message Sender" to push messages to everyone who is subscribed to a message topic through our other API modules.

推荐答案

所以第二种方法确实有效.我将 websockets 配置为独立运行(无中继)，然后我在服务层建立了一个单独的 AMQP 消息代理连接，以允许服务之间进行通信.在 API 模块中，我只是监听 AMQP 消息代理，然后手动将这些消息转发到通知 websocket 订阅者的 SimpMessagingTemplate.我不确定这在技术上是否是正确"的方法，但它似乎工作得很好，我还没有看到任何实施问题.事实上，我实际上认为我可能更喜欢这种方法，因为我现在只是让我的所有服务能够通过比我最初需要的 websockets 更多类型的消息来相互交谈.

So the second approach did in fact work. I configured the websockets to be run independently (no relay) and then I made a separate AMQP message broker connection at the service layer to allow communication between services. In the API module, I simply listened to the AMQP message broker and then manually forwarded those messages to the SimpMessagingTemplate which notified the websocket subscribers. I am not sure if this is technically the "right" way to do it but it seems to be working great and I do not yet see any issues with the implementation. In fact, I actually think I may prefer this approach as I now just gave all my services the ability to talk to each other with more types of messages than what I originally needed for the websockets.

这是新的配置:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer
{
    @Override
    public void configureMessageBroker(final MessageBrokerRegistry config)
    {
        config.enableSimpleBroker("/topic");
    }

    @Override
    public void registerStompEndpoints(final StompEndpointRegistry registry)
    {
        registry.addEndpoint("/updates")
            .setAllowedOrigins("*")
            .withSockJS();
    }
}

这里是我监听消息代理并将消息转发给 websocket 订阅者的地方:

And here is where I listen to the message broker and forward the messages to the websocket subscribers:

@Component
public class SendWebSocketUpdates
{
    private static final Logger logger = LoggerFactory.getLogger(SendWebSocketUpdates.class);

    private final Gson gson;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private MessageBrokerConsumer<String> messageBrokerConsumer;

    public SendWebSocketUpdates()
    {
        this.gson = new Gson();
    }

    @PostConstruct
    public void init()
    {
        //listen for incoming AMQP messages from the rabbitmq server and forward them to the websocket subscribers
        messageBrokerConsumer.addListener((message, topicName) -> {
            final String destination = "/topic/" + topicName;
            final String messageJson = gson.toJson(message.getBody());

            //check to see if trace logging is enabled
            if (logger.isTraceEnabled())
            {
                logger.trace("Sending Message to \"{}\": {}", destination, messageJson);
            }

            //broadcast the via a STOMP message to subscribers of this topic
            messagingTemplate.convertAndSend(destination, messageJson);
        });
    }
}

这篇关于从应用程序的其他层发送 STOMP 消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！