Spring Integration channel()消息分发至意外订阅者问题排查

使用Spring Integration Flows的channel()方法时出现异常，不确定是理解错误还是框架Bug。

问题代码 #

@Configuration
@RequiredArgsConstructor
@FieldDefaults(makeFinal = true, level = AccessLevel.PRIVATE)
public class ImportFlow {

    ConsumerFactory<String, String> kafkaConsumerFactory;
    ServiceProperties serviceProperties;

    DebugHandler debugHandler;
    DebugSecondHandler secondHandler;
    TombstoneHandler tombstoneHandler;
    UnexpectedHandler unexpectedHandler;

    @Bean
    public StandardIntegrationFlow startKafkaInbound() {
        return IntegrationFlow.from(Kafka
                .messageDrivenChannelAdapter(
                    kafkaConsumerFactory,
                    ListenerMode.record,
                    serviceProperties.getImportTopic().getName())
            )
            .channel(Objects.requireNonNull(routeKafkaMessage().getInputChannel()))
            .get();
    }

    @Bean
    public IntegrationFlow routeKafkaMessage() {
        return IntegrationFlow.from("routeKafkaMessage.input")
            .route(Message.class,
                p -> {
                    if (p.getPayload().equals(KafkaNull.INSTANCE)) {
                        return processTombstoneFlow().getInputChannel();
                    }
                    return someSplittingFlow().getInputChannel();
                })
            .get();
    }

    @Bean
    IntegrationFlow someSplittingFlow() {
        return IntegrationFlow.from("someSplittingFlow.input")
            .handle(debugHandler)
            .channel(sendToFlow().getInputChannel())
            .get();
    }

    @Bean
    IntegrationFlow processTombstoneFlow() {
        return IntegrationFlow.from("processTombstoneFlow.input")
            .handle(tombstoneHandler)
            .channel(sendToFlow().getInputChannel())
            .handle(unexpectedHandler) // 此处除直接get()外的操作都会引发错误
            // .nullChannel()
            .handle(m -> {})
            .get();
    }

    @Bean
    IntegrationFlow sendToFlow() {
        return IntegrationFlow.from("sendToFlow.input")
            .handle(secondHandler)
            .handle(m -> {}, e -> e.id("endOfSendToFlow"))
            .get();
    }
}

异常现象 #

• 向Kafka发送100条带Payload的消息，仅50条被sendToFlow中的secondHandler接收，另外50条被unexpectedHandler接收，预期所有消息都应发送至sendToFlow。
• 日志显示每隔一条消息被发送至tombstoneFlow：

org.springframework.integration.channel.DirectChannel: preSend on channel 'bean 'processTombstoneFlow.channel#0'

• 发送100条KafkaNull消息时，仅半数到达secondHandler。
• 预期someSplittingFlow和processTombstoneFlow分别通过独立DirectChannel连接sendToFlow，但两者产生了关联；新增第三个通道向sendToFlow发送消息时，仅33%的消息到达预期通道。

问题根源 #

在路由逻辑和子Flow中，直接调用processTombstoneFlow().getInputChannel()、someSplittingFlow().getInputChannel()等Bean方法，会导致每次调用都创建新的IntegrationFlow实例，而非复用Spring容器中已托管的Bean：

• 每次路由判断时调用processTombstoneFlow()，都会生成新的IntegrationFlow Bean（包含新的processTombstoneFlow.input通道和后续处理链）。
• 同理，someSplittingFlow()和sendToFlow()每次调用也会创建新实例，导致容器中存在多个同名通道（如sendToFlow.input）。DirectChannel默认采用轮询负载均衡策略，消息会被分发到不同的Flow实例，最终出现消息分流的异常现象。

解决方案 #

1. 通过通道名称关联Flow，避免直接调用Bean方法：
   路由时直接返回通道名称，子Flow之间通过通道名称连接，确保复用Spring容器中唯一的通道实例：

   @Bean
   public IntegrationFlow routeKafkaMessage() {
       return IntegrationFlow.from("routeKafkaMessage.input")
           .route(Message.class, p -> {
               if (p.getPayload().equals(KafkaNull.INSTANCE)) {
                   return "processTombstoneFlow.input";
               }
               return "someSplittingFlow.input";
           })
           .get();
   }
   
   @Bean
   IntegrationFlow someSplittingFlow() {
       return IntegrationFlow.from("someSplittingFlow.input")
           .handle(debugHandler)
           .channel("sendToFlow.input")
           .get();
   }
   
   @Bean
   IntegrationFlow processTombstoneFlow() {
       return IntegrationFlow.from("processTombstoneFlow.input")
           .handle(tombstoneHandler)
           .channel("sendToFlow.input")
           // 移除多余的handle操作，避免消息被额外处理
           .get();
   }
   

2. 注入通道实例替代调用Bean方法：
   在ImportFlow中注入所需的通道Bean，确保使用容器中唯一的实例：

   @Qualifier("routeKafkaMessage.input")
   MessageChannel routeInputChannel;
   
   @Bean
   public StandardIntegrationFlow startKafkaInbound() {
       return IntegrationFlow.from(Kafka
               .messageDrivenChannelAdapter(
                   kafkaConsumerFactory,
                   ListenerMode.record,
                   serviceProperties.getImportTopic().getName())
           )
           .channel(routeInputChannel)
           .get();
   }
   

3. 清理冗余处理逻辑：
   移除processTombstoneFlow中多余的handle(unexpectedHandler)和handle(m -> {})，避免消息被额外处理。

内容的提问来源于stack exchange，提问作者Jonathan H