将spring-reactor集成到现有的Spring Framework 4 STOMP Over

Question

@Bean
public ThreadPoolTaskExecutor clientInboundChannelExecutor() {
    TaskExecutorRegistration reg = getClientInboundChannelRegistration().getOrCreateTaskExecRegistration();
    ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
    executor.setThreadNamePrefix("clientInboundChannel-");
    return executor;
}

当我尝试在WebSocketConfig中重写此方法时，“来自ChannelRegistration类型的方法getOrCreateTaskExecRegistration（）不可见”，因为在AbstractMessageBrokerConfiguration中它受到保护....

protected final ChannelRegistration getClientInboundChannelRegistration() {
    if (this.clientInboundChannelRegistration == null) {
        ChannelRegistration registration = new ChannelRegistration();
        configureClientInboundChannel(registration);
        this.clientInboundChannelRegistration = registration;
    }
    return this.clientInboundChannelRegistration;
}

我不完全理解WebSocketMessageBrokerConfigurationSupport层次结构或WebSocketMessageBrokerConfigurer接口。我只是玩弄重写我的定制工作所需的东西。

不确定它是否相关，但我不需要外部代理，因为我的应用程序目前不向所有连接的订阅者发送任何数据，而且不太可能最终发送数据。与守护进程类型的java websocket客户机的通信是点对点的，但是浏览器中的web ui websocket确实使用subscribe来获取实时数据，因此这是一个方便的设置（而不是spring integration direct channel)，并且有关于如何设置的明确来源--但我不确定这是最有效的应用程序设计。spring-framework参考文档中描述的WebSocket消息传递体系结构上的STOMP是最全面的方法，因为这是我的第一个spring项目。

我应该如何将spring-reactor集成到我的标准spring framework 4 STOMP Over WebSocket消息传递架构中？

如果在clientInboundChannelExecutor和clientOutboundChannelExecutor中配置RingBufferAsyncTaskExecutor代替ThreadPoolTaskExecutor是正确的方法，那么应该如何执行呢？

Answer 1

实际上RingBufferAsyncTaskExecutor不是ThreadPoolTaskExecutor，所以不能这样使用它。

您可以简单地从AbstractWebSocketMessageBrokerConfigurerimpl重写ClientInbound（出站）通道beans，只需使用@enableWebSocketMessageBroker:

@Configuration
@EnableWebSocketMessageBroker
@EnableReactor
public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {

    @autowired
    Environment reactorEnv;

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry configurer) {
        configurer.setApplicationDestinationPrefixes("/app");
        configurer.enableSimpleBroker("/topic", "/queue");
    }

    @Bean
    public AbstractSubscribableChannel clientInboundChannel() {
        ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(new RingBufferAsyncTaskExecutor(this.reactorEnv));
        ChannelRegistration reg = getClientOutboundChannelRegistration();
        channel.setInterceptors(reg.getInterceptors());
        return channel;
    }

}

请注意Spring集成中的WebSocket支持。