执行器通道中的线程安全

Question

消费者是一个spring集成项目，它从消息队列中消费并执行大量处理。我使用Executor通道并行处理消息，然后流通过一些公共处理程序类。

请在下面的代码片段中找到-

1. baseVentFlow（）-我们从EMS队列接收消息并将其发送到路由器
2. router()-基于以下消息的id：“特定ExecutorChannel实例配置了一个单线程执行器。每个ExecutorChannel都将是它的专用执行器，只有一个线程。
3. skwdefaultChannel()、gjsucaDefaultChannel()、rpaDefaultChannel()-所有ExecutorChannel bean都用@bridgeto标记为启动公共流的同一通道。
4. uaEventFlow（）-这里将处理每个消息

@Bean
public IntegrationFlow baseEventFlow() {
    return IntegrationFlows
            .from(Jms.messageDrivenChannelAdapter(Jms.container(this.emsConnectionFactory, this.emsQueue).get()))
            .wireTap(FLTAWARE_WIRE_TAP_CHNL)
            .route(router()).get();
}

public AbstractMessageRouter router() {
    return new AbstractMessageRouter() {
        @Override
        protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
            if (message.getPayload().toString().contains("\"id\":\"RPA")) {
                return Collections.singletonList(skwDefaultChannel());
            }else if (message.getPayload().toString().contains("\"id\":\"ASH")) {
                return Collections.singletonList(rpaDefaultChannel());
            } else if (message.getPayload().toString().contains("\"id\":\"GJS")
                    || message.getPayload().toString().contains("\"id\":\"UCA")) {
                return Collections.singletonList(gjsucaDefaultChannel());
            } else {
                return Collections.singletonList(new NullChannel());
            }
        }
    };
}

@Bean
@BridgeTo("uaDefaultChannel")
public MessageChannel skwDefaultChannel() {
    return MessageChannels.executor(SKW_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();
}

@Bean
@BridgeTo("uaDefaultChannel")
public MessageChannel gjsucaDefaultChannel() {
    return MessageChannels.executor(GJS_UCA_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();
}

@Bean
@BridgeTo("uaDefaultChannel")
public MessageChannel rpaDefaultChannel() {
    return MessageChannels.executor(RPA_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();
}

@Bean
public IntegrationFlow uaEventFlow() {
    return IntegrationFlows.from("uaDefaultChannel")
             .wireTap(UA_WIRE_TAP_CHNL)
             .transform(eventHandler, "parseEvent")
             .handle(uaImpl, "process").get();
}

我担心的是在uaEVentFlow（）中，常见的转换和处理程序方法不是线程安全的，这可能会导致问题。我们如何确保在每次消息调用时注入一个新的转换器和处理程序？我应该将转换器和处理程序bean的范围改为原型吗？

Answer 1

您应该将.transform()和.handle()移动到每个上游流，并添加

java prettyprint-override">@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

到它们的@bean定义中，以便每个都获得自己的实例。

但是，通常最好使代码是线程安全的。