多线程执行器通道，加快使用者进程

Question

消费者是一个spring集成项目，它从消息队列中消费并执行大量处理。目前，它是单线程的，不能与生产者发送消息的速度匹配。因此队列深度不断增加

        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(Jms.container(this.emsConnectionFactory, this.emsQueue).get()))
                .wireTap(FLTAWARE_WIRE_TAP_CHNL)// push raw fa data
                .filter(ingFilter, "filterMessageOnEvent").transform(eventHandler, "parseEvent")
                .aggregate(a -> a.correlationStrategy(corStrgy, "getCorrelationKey").releaseStrategy(g -> {

                    boolean eonExists = g.getMessages().stream()
                            .anyMatch(eon -> ((FlightModel) eon.getPayload()).getEstGmtOnDtm() != null);
                    if (eonExists) {
                        boolean einExists = g.getMessages().stream()
                                .anyMatch(ein -> ((FlightModel) ein.getPayload()).getEstGmtInDtm() != null);
                        if (einExists) {
                            return true;
                        }
                    }
                    return false;
                }).messageStore(this.messageStore)).channel("AggregatorEventChannel").get();

是否可以使用executor通道在多线程环境中处理此问题并加快使用者进程

如果是，请建议如何实现--为了确保消息的顺序，我需要将相同类型的消息（基于消息的id）分配给executor通道的相同线程。

    public static final MessageChannel SKW_DEFAULT_CHANNEL = MessageChannels
            .executor(ASQ_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();

    public static final MessageChannel RPA_DEFAULT_CHANNEL = MessageChannels
            .executor(ASH_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();

    @Bean
    public IntegrationFlow baseEventFlow1() {

        return IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(Jms.container(this.emsConnectionFactory, this.emsQueue).get()))
                .wireTap(FLTAWARE_WIRE_TAP_CHNL)// push raw fa data
                .filter(ingFilter, "filterMessageOnEvent").route(route()).get();
    }

    public AbstractMessageRouter router() {
        return new AbstractMessageRouter() {
            @Override
            protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
                if (message.getPayload().toString().contains("\"id\":\"RPA")) {
                    return Collections.singletonList(RPA_DEFAULT_CHANNEL);
                } else if (message.getPayload().toString().contains("\"id\":\"SKW")) {
                    return Collections.singletonList(SKW_DEFAULT_CHANNEL);
                } else {
                    return Collections.singletonList(new NullChannel());
                }
            }

        };
    }

    @Bean
    @BridgeTo("uaxDefaultChannel")
    public MessageChannel ucaDefaultChannel() {
        return MessageChannels.executor(UCA_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();
    }

    @Bean
    @BridgeTo("uaDefaultChannel")
    public MessageChannel ualDefaultChannel() {
        return MessageChannels.executor(UAL_DEFAULT_CHANNEL_NAME, Executors.newFixedThreadPool(1)).get();
    }
    @Bean
    public IntegrationFlow uaEventFlow() {
        return IntegrationFlows.from("uaDefaultChannel").wireTap(UA_WIRE_TAP_CHNL)
                .transform(eventHandler, "parseEvent")
}

因此executor通道上的BridgeTo将转发消息

Answer 1

因此队列深度不断增加

由于看起来您的队列位于JMS broker上的某个位置，所以确实可以有这样的行为。这正是为消息传递系统所设计的--区分生产者和消费者，并在可能的情况下处理目的地的消息。

如果希望增加来自JMS的轮询，可以考虑在JMS容器上设置concurrency选项：

/**
 * The concurrency to use.
 * @param concurrency the concurrency.
 * @return current {@link JmsDefaultListenerContainerSpec}.
 * @see DefaultMessageListenerContainer#setConcurrency(String)
 */
public JmsDefaultListenerContainerSpec concurrency(String concurrency) {
    this.target.setConcurrency(concurrency);
    return this;
}

/**
 * The concurrent consumers number to use.
 * @param concurrentConsumers the concurrent consumers count.
 * @return current {@link JmsDefaultListenerContainerSpec}.
 * @see DefaultMessageListenerContainer#setConcurrentConsumers(int)
 */
public JmsDefaultListenerContainerSpec concurrentConsumers(int concurrentConsumers) {
    this.target.setConcurrentConsumers(concurrentConsumers);
    return this;
}

/**
 * The max for concurrent consumers number to use.
 * @param maxConcurrentConsumers the max concurrent consumers count.
 * @return current {@link JmsDefaultListenerContainerSpec}.
 * @see DefaultMessageListenerContainer#setMaxConcurrentConsumers(int)
 */
public JmsDefaultListenerContainerSpec maxConcurrentConsumers(int maxConcurrentConsumers) {
    this.target.setMaxConcurrentConsumers(maxConcurrentConsumers);
    return this;
}