java - 频繁发送到spring-websocket session : lost in transit

Question

我得到了 spring websocket 服务器(基于 Jetty 和 spring 版本 4.3.2.RELEASE)和客户端的负载测试设置，它生成许多连接(基于 spring 的示例 java websocket 客户端)。下面的代码将数据发送到给定的 websocket session :该代码片段利用了可以使用 sessionId 而不是用户 ID ( Spring WebSocket @SendToSession: send message to specific session ) 的情况。我可能会经常执行此代码，每 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());

}

当此代码非常频繁地(每 2-3 毫秒)执行约 100 个 session 时，虽然我在日志中看到它已运行并调用了 ConvertAndSendToUser，但某些 session 将不会收到消息。我感谢任何有关如何清除此问题的建议。

Answer 1

嗯，我认为你的问题在于:

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

它在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时，任何多余的都会被中止。

要解决此问题，您应该实现 WebSocketMessageBrokerConfigurer 并重写其 configureClientOutboundChannel() 以提供一些自定义 taskExecutor(ThreadPoolTask​​Executor taskExecutor)，例如使用 new ThreadPoolExecutor.CallerRunsPolicy()。