Spring Integration中存在PubSub适配器时flowA的enrichHeaders未执行

问题出在DirectChannel的点对点特性以及你共享了channelA作为两个不同生产者的目标通道，同时它也是flowA的输入通道。

当你将PubSubInboundChannelAdapter的outputChannel设置为channelA（DirectChannel）时，Spring Integration初始化过程中会将适配器的消息发送逻辑与channelA绑定，导致flowA的完整处理器链（包含enrichHeaders步骤）无法正确注册为channelA的唯一消费者。此时，通过flowHttp发送到channelA的消息会绕过enrichHeaders，直接流向后续的filter和handle环节。

当你注释掉PubSubInboundChannelAdapter时，flowA的处理器链会正常注册为channelA的唯一消费者，所有发送到channelA的消息都会完整经过enrichHeaders步骤。

方法1：避免共享DirectChannel，使用独立输入通道 #

不要手动定义channelA，让Spring Integration为flowA自动创建输入通道，同时将PubSubInboundChannelAdapter的输出通道指向这个自动创建的通道：

@Bean
public PubSubInboundChannelAdapter pubsubInboundChannelAdapter(
        IntegrationFlow flowA,
        PubSubTemplate pubSubTemplate) {
    PubSubInboundChannelAdapter adapter = 
        new PubSubInboundChannelAdapter(pubSubTemplate, subscriptionName);
    adapter.setOutputChannel(flowA.getInputChannel());
    adapter.setAckMode(AckMode.AUTO_ACK);
    adapter.setPayloadType(String.class);
    adapter.setErrorChannelName("errorChannel");
    return adapter;
}

@Bean
public IntegrationFlow flowA(Filter1 filter1, Service1 service1) {
    return IntegrationFlow.from(MessageChannels.direct())
                .enrichHeaders(spec -> spec.header("flowName", "flowA", true))
                .filter(filter1, "filterIt1")
                .handle(service1, "handleIt1")
                .get();
}

方法2：将channelA改为PublishSubscribeChannel #

如果业务允许，将channelA改为支持多消费者的PublishSubscribeChannel，这样它可以同时接收两个生产者的消息并转发给flowA的处理器链：

@Bean
public MessageChannel channelA() {
    return new PublishSubscribeChannel();
}

方法3：在flowHttp中显式复用flowA的逻辑 #

这种方法会避免通道共享问题，但会重复代码，适合简单场景：

@Bean
public IntegrationFlow flowHttp(Filter1 filter1, Service1 service1) {
    return IntegrationFlow.from(Http.inboundChannelAdapter("/messageA")
                        .requestMapping(m -> m.methods(HttpMethod.POST).consumes("application/json"))
                        .payloadExpression("body")
                        .requestPayloadType(String.class))
                .enrichHeaders(spec -> spec.header("flowName", "flowA", true))
                .filter(filter1, "filterIt1")
                .handle(service1, "handleIt1")
                .get();
}

内容的提问来源于stack exchange，提问作者Shinta Smith