Spring Integration队列通道消息如何即时触发Service Activator

问题根因 #

QueueChannel属于异步缓冲通道，消息投递后会暂存在内部队列中，必须通过轮询器主动拉取才能触发Service Activator执行，你遇到的触发延迟本质是轮询配置未达到最优。 #

解决方案 #

方案1：优化轮询配置（保留QueueChannel异步特性的前提下实现近即时触发）

核心是将轮询器的receive-timeout设为-1，此时轮询线程会阻塞在队列的读取操作上，直到有新消息到达立刻触发处理，不会产生空轮询间隔。

XML配置版本

可以选择优化全局轮询器，或者给目标Service Activator配置专用轮询器避免被其他组件影响：

<!-- 优化全局轮询器 -->
<int:poller default="true" fixed-delay="0" max-messages-per-poll="1" receive-timeout="-1"/>

<!-- 或单独配置专用轮询器 -->
<int:service-activator input-channel="demoInputChannel"
                       output-channel="demoOutputChannel"
                       ref="demoService"
                       method="demoMethod"
                       requires-reply="true">
    <int:poller fixed-delay="0" max-messages-per-poll="1" receive-timeout="-1"/>
</int:service-activator>

注解配置版本

首先定义专用轮询器Bean，再给Service Activator绑定该轮询器：

// 定义即时轮询器
@Bean
public PollerMetadata immediatePoller() {
    PollerMetadata poller = new PollerMetadata();
    poller.setFixedDelay(0);
    poller.setMaxMessagesPerPoll(1);
    poller.setReceiveTimeout(-1);
    return poller;
}

// 绑定轮询器到Service Activator
@ServiceActivator(inputChannel = "demoInputChannel", 
                  outputChannel = "demoOutputChannel",
                  poller = @Poller("immediatePoller"))
public Message<List<DataElement>> handle(Message<Long> in) throws Exception {
    System.out.println("item.getPayload " + in.getPayload());
    return autowiredObject.autowiredObjectMethod(in);
}

如果消费速度跟不上发送速度导致队列堆积，可给轮询器增加线程池配置，实现多线程并行消费：

@Bean
public PollerMetadata immediatePoller() {
    PollerMetadata poller = new PollerMetadata();
    poller.setFixedDelay(0);
    poller.setMaxMessagesPerPoll(1);
    poller.setReceiveTimeout(-1);
    // 配置消费线程池
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(0);
    executor.initialize();
    poller.setTaskExecutor(executor);
    return poller;
}

方案2：替换为DirectChannel（追求最高即时性的最优解）

如果不需要QueueChannel的异步缓冲、削峰特性，直接将通道类型改为DirectChannel即可，消息投递后会直接在发送线程同步调用Service Activator，完全没有延迟，也不需要配置轮询器：

@Bean
MessageChannel demoInputChannel() {
    return new DirectChannel();
}

额外检查项 #

确认Splitter的输出通道配置正确，没有出现消息发错通道的情况，你当前的Splitter配置逻辑本身没有问题。

内容的提问来源于stack exchange，提问作者İlkay Gunel