频繁发送到spring websocket会话：在传输过程中丢失

Question

我得到了一个SpringwebSocket服务器的负载测试设置（基于Jetty和Spring版本4.3.2。RELEASE）和客户端，生成许多连接（基于Spring的示例java webSocket客户端）。下面的代码将数据发送到给定的webSocket会话：代码段利用了可以使用sesionId而不是User ID的情况（Spring WebSocket@SendTo会话：将消息发送到特定会话）。我可能会非常频繁地执行这段代码，每2-3毫秒执行一次。我使用SimpleMessageBroker。

 public void publishToSessionUsingTopic(String sessionId, String subscriptionTopic, Map<String, CacheRowModel> payload) {

        String subscriptionTopicWithoutUser = subscriptionTopic.replace(USER_ENDPOINT, "");
        // necessary message headers for per-session send
        SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
        headerAccessor.setSessionId(sessionId);
        headerAccessor.setLeaveMutable(true);          
        simpMessagingTemplate.convertAndSendToUser(sessionId, subscriptionTopicWithoutUser, Collections.singletonList(payload), headerAccessor.getMessageHeaders());

}

当这段代码在大约100个会话中非常频繁地执行（每2-3毫秒）时，虽然我在日志中看到它被运行并被称为转换器和SendToUser，但有些会话不会收到消息。我感谢任何关于如何清理这一问题的建议。

Answer 1

我认为你的问题在于：

@Bean
public ThreadPoolTaskExecutor clientOutboundChannelExecutor() {
    TaskExecutorRegistration reg = getClientOutboundChannelRegistration().getOrCreateTaskExecRegistration();
    ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
    executor.setThreadNamePrefix("clientOutboundChannel-");
    return executor;
}

其中，它将此配置用于执行器：

protected ThreadPoolTaskExecutor getTaskExecutor() {
    ThreadPoolTaskExecutor executor = (this.taskExecutor != null ? this.taskExecutor : new ThreadPoolTaskExecutor());
    executor.setCorePoolSize(this.corePoolSize);
    executor.setMaxPoolSize(this.maxPoolSize);
    executor.setKeepAliveSeconds(this.keepAliveSeconds);
    executor.setQueueCapacity(this.queueCapacity);
    executor.setAllowCoreThreadTimeOut(true);
    return executor;
}

请参阅，未配置RejectedExecutionHandler。默认情况下，它类似于：

private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

因此，当您有足够多的消息和任务超过ThreadPool时，任何额外的消息和任务都将被中止。

要解决此问题，您应该实现websocketmessagebrokerconfigure并覆盖其configureClientOutboundChannel（），以提供一些自定义任务执行器（ThreadPoolTaskExecutor taskExecutor），例如使用新的ThreadPoolExecutor。CallerRunPolicy（）。