RabbitMQ Stomp无法获得连续的大型消息

Question

我正在使用带有Stomp的Spring RabbitMQ来构建一个事件流应用程序，其中服务器将大约100KB的大消息连续地放到TopicExchange amp.topic，绑定密钥为test，并且stomp客户端订阅了/topic/test。

这是代码：

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

private static final int PORT = 1234;

/**
 * Methos to configure the Message Broker, in this case StompMessageBroker
 * @param config
 */
@Override
public void configureMessageBroker(MessageBrokerRegistry config) {

     config.enableStompBrokerRelay("/topic", "/queue/")
             .setRelayHost(host)
             .setRelayPort(PORT)
             .setClientLogin(username)
             .setClientPasscode(password)      
     config.setApplicationDestinationPrefixes("/app");
}

/**
 * Method to configure Stomp endpoints used by the client to connect using websockets.
 * @param registry
 */
@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/test-websocket")
            .setAllowedOrigins("*").withSockJS();
  }
}

交换信息：

public static final String EXCHANGE_NAME="amq.topic";
// This call is done in a loop 100 times and puts 500KB of message each time to exchange.
 Rabbittemplate.convertAndSend(EXCHANGE_NAME, "test_user", "Loop Counter_"+i+" : " + str); // str is a string message of size 500KB.

我的前端应用程序使用：

sockJs = new SockJS('test-websocket');
stompClient = Stomp.over(sockJs);
stompClient.connect('username','password', (frame: any)=> {

var subscription_id =   this.stompClient1.subscribe('/topic/test_user',  (greeting: any) => {
      var message_id = greeting.headers['message-id'];
      stompClient.ack(message_id, subscription_id);
  },{ack: 'client'});
});

Q1。当我发送更大的消息（每个> 100KB）以循环交换（连续100次）时，stomp客户端无法获得控制台消息

Whoops! Lost connection to test-websocket

任何帮助表示赞赏。

Answer 1

这可能是由websocket队列的超出缓冲区大小引起的。你发送的websocket太快了。尝试在spring-websocket传输配置上增加websocket缓冲区大小：

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketMessageBrokerConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
        registry.setSendBufferSizeLimit(100 * 1024 * 1024); // 100MB
        registry.setSendTimeLimit(60_000);
    }

}

但在此之前，为DEBUG记录器启用org.springframework.web.socket.messaging级别日志记录，您应该看到如下消息：

2019-04-08 19:16:40.171 DEBUG 15428 --- [clientOutboundChannel-5] o.s.w.s.m.SubProtocolWebSocketHandler    : Terminating 'WebSocketServerSockJsSession[id=f46296cfb74949b3af04b53739dc83ba]'

org.springframework.web.socket.handler.SessionLimitExceededException: Buffer size 524450 bytes for session 'f46296cfb74949b3af04b53739dc83ba' exceeds the allowed limit 524288
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.limitExceeded(ConcurrentWebSocketSessionDecorator.java:227) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.checkSessionLimits(ConcurrentWebSocketSessionDecorator.java:197) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.sendMessage(ConcurrentWebSocketSessionDecorator.java:150) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.StompSubProtocolHandler.sendToClient(StompSubProtocolHandler.java:455) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.StompSubProtocolHandler.handleMessageToClient(StompSubProtocolHandler.java:442) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.web.socket.messaging.SubProtocolWebSocketHandler.handleMessage(SubProtocolWebSocketHandler.java:355) ~[spring-websocket-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:144) [spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_152]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_152]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_152]

这清楚地描述了这个问题。