执行器通道中的线程安全答案

【问题标题】：Thread safety in executor channel执行器通道中的线程安全
【发布时间】：2020-05-17 11:34:23
【问题描述】：

我有一个消息生成器，它每秒产生大约 15 条消息

consumer 是一个 spring 集成项目，它从 Message Queue 中消费并进行大量处理。我使用 Executor 通道并行处理消息，然后流通过一些通用处理程序类。

请在sn-p下面找到代码-

baseEventFlow() - 我们从 EMS 队列接收消息并将其发送到路由器
router() - 基于消息的 id”，一个特定的 ExecutorChannel 实例配置了一个单线程的 Executor。每个 ExecutorChannel 都将是它的专用 executor，只有一个线程。
skwDefaultChannel(), gjsucaDefaultChannel(), rpaDefaultChannel() - 所有 ExecutorChannel bean 都用 @BridgeTo 标记，用于启动该公共流程的同一通道。
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 的范围更改为原型吗？

【问题讨论】：

标签： spring-integration spring-integration-dsl

【解决方案1】：

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

@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

到他们的@Bean 定义，所以每个都有自己的实例。

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

【讨论】：