如何基于订阅而非轮询用Spring Integration推送BlockingQueue消息至动态WebSocket

当然有更优的方案——利用BlockingQueue的阻塞特性，让消息源仅在队列有消息时才触发处理，彻底替代轮询机制，大幅提升效率。

核心优化思路 #

摒弃定时轮询的方式，让消息源线程在队列无消息时挂起，有新消息时自动唤醒处理，完全消除无效轮询带来的资源开销。

具体实现代码 #

替换原有的MethodInvokingMessageSource和固定速率轮询配置，改为基于BlockingQueue的阻塞式消息源：

@Service
public class WebSocketPublisherService {

    @Autowired
    private IntegrationFlowContext integrationFlowContext;

    @Bean
    public HandshakeHandler handshakeHandler() {
        return new DefaultHandshakeHandler(new TomcatRequestUpgradeStrategy());
    }

    // 业务阻塞队列，可根据实际需求注入或初始化
    private final BlockingQueue<Object> businessMessageQueue = new LinkedBlockingQueue<>();

    public void createWebSocket() {
        ServerWebSocketContainer serverWebSocketContainer = new ServerWebSocketContainer("/test")
                .setHandshakeHandler(handshakeHandler())
                .setAllowedOrigins("http://localhost:4200");
        serverWebSocketContainer.setMessageListener(session -> {
            // 保留处理前端WebSocket消息的逻辑
        });

        WebSocketOutboundMessageHandler webSocketOutboundMessageHandler = 
            new WebSocketOutboundMessageHandler(serverWebSocketContainer);

        // 基于BlockingQueue创建阻塞式消息源
        MessageSource<Object> blockingQueueSource = () -> {
            try {
                // take()会阻塞直到队列有新消息
                Object payload = businessMessageQueue.take();
                return MessageBuilder.withPayload(payload).build();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
        };

        // 构建IntegrationFlow，配置阻塞式轮询器
        StandardIntegrationFlow standardIntegrationFlow = IntegrationFlow
                .from(blockingQueueSource, polling -> polling.poller(pollerFactory -> 
                    // receiveTimeout(-1)表示无限等待，直到队列有消息
                    pollerFactory.receiveTimeout(-1)
                ))
                .split(new DecorateMessageWithSessionId(serverWebSocketContainer))
                .handle(webSocketOutboundMessageHandler)
                .get();

        IntegrationFlowContext.IntegrationFlowRegistration flowRegistration = integrationFlowContext
                .registration(standardIntegrationFlow)
                .addBean(serverWebSocketContainer)
                .register();
        flowRegistration.start();
    }
}

关键优化点说明 #

• 阻塞式消息获取：用BlockingQueue.take()替代轮询取数，线程在队列空时自动挂起，有新消息立即唤醒处理，彻底消除无效轮询开销。
• Poller配置调整：将原有的fixedRate(10)改为receiveTimeout(-1)，让Spring Integration的轮询器阻塞等待消息，而非定时触发。
• 可选规范实现：也可以用QueueChannel封装业务队列（QueueChannel queueChannel = new QueueChannel(businessMessageQueue)），直接作为消息源传入IntegrationFlow.from()，内部已封装阻塞式接收逻辑，代码更简洁。

额外注意事项 #

• 向队列添加消息时，直接调用businessMessageQueue.put(message)即可，消息会自动触发Flow的处理流程。
• 动态创建的IntegrationFlow需要在合适时机销毁（比如WebSocket连接全部关闭时），可通过flowRegistration.destroy()释放资源。
• 确保DecorateMessageWithSessionId能正确为消息添加目标WebSocket会话ID的头信息，让WebSocketOutboundMessageHandler精准推送消息到对应客户端。

内容的提问来源于stack exchange，提问作者Andrei Roșu-Cojocaru