Thread safety in executor channel

Question

I have a message producer which produces around 15 messages/second

The consumer is a spring integration project which consumes from the Message Queue and does a lot of processing. I have used the Executor channel to process messages in parallel and then the flow passes through some common handler class.

Please find below the snippet of code -

baseEventFlow() - We receive the message from the EMS queue and send it to a router
router() - Based on the id of the message" a particular ExecutorChannel instance is configured with a singled-threaded Executor. Every ExecutorChannel is going to be its dedicated executor with only single thread.
skwDefaultChannel(), gjsucaDefaultChannel(), rpaDefaultChannel() - All the ExecutorChannel beans are marked with the @BridgeTo for the same channel which starts that common flow.
uaEventFlow() - Here each message will get processed

@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();
}

My concern is in the uaEVentFlow() the common transform and handler method are not thread safe and it may cause issue. How can we ensure that we inject a new transformer and handler at every message invocation? Should I change the scope of the transformer and handler bean as prototype?

Gary Russell Gary Russell · Accepted Answer · 2020-05-17T16:22:38

Instead of bridging to a common flow, you should move the .transform() and .handle() to each of the upstream flows and add

@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)

to their @Bean definitions so each gets its own instance.

But, it's generally better to make your code thread-safe.

Thread safety in executor channel

1 Answers