Spring Integration Java DSL中通知即忘模式的正确实现方式

首选方案：Wire Tap（精准匹配需求） #

Wire Tap就是Spring Integration专门为不中断主流程、异步/同步复制消息到子流程做独立处理设计的组件，你之前遇到的问题应该是子流程配置有误（比如误操作了原消息而非副本）。正确使用时，子流程的所有操作（包括过滤、转换）只会作用于复制的消息副本，完全不影响主流程的原消息流转。

对应你业务的代码示例 #

主流程定义

@Bean
public IntegrationFlow mainFlow() {
    return IntegrationFlows.from(gateway()) // 从网关获取消息
            .transform(xmlToDtoTransformer()) // XML转DTO
            .filter(message -> /* 系统条件过滤逻辑 */, 
                    spec -> spec.discardChannel("discardChannel")) // 不满足则终止消息
            // 触发反馈子流程：仅部分消息参与，子流程内做过滤
            .wireTap(feedbackSubFlow())
            // 主流程继续处理原消息
            .transform(dtoSerializer()) // 消息序列化
            .handle(hmacSignHandler()) // HMAC签名
            // 触发超时记录子流程：即忘式处理
            .wireTap(timeoutRecordSubFlow())
            .channel("mainChannel") // 最终发送到IBM MQ通道
            .get();
}

反馈子流程（内部做参与条件过滤）

@Bean
public IntegrationFlow feedbackSubFlow() {
    return f -> f
            .filter(message -> /* 反馈子流程参与条件 */) // 仅符合条件的消息进入
            .transform(feedbackDtoGenerator()) // 生成反馈消息DTO
            .serialize() // 序列化反馈消息
            .channel("feedbackChannel"); // 发送到JMS转AMQP通道
}

超时记录子流程

@Bean
public IntegrationFlow timeoutRecordSubFlow() {
    return f -> f
            .transform(message -> /* 提取消息ID等核心信息 */)
            .handle(timeoutConsumerHandler()); // 发送到消费者，无需回复
}

关键优化点 #

• 如果想让子流程异步执行、不阻塞主流程，可以给Wire Tap指定线程池通道：

.wireTap(f -> f.channel(MessageChannels.executor(Executors.newCachedThreadPool()))
               .handle(feedbackSubFlow()))

• 若你的DTO是可变对象，子流程修改DTO会影响主流程的话，可在Wire Tap时先做深拷贝：

.wireTap(f -> f.transform(message -> {
    Dto original = (Dto) message.getPayload();
    Dto copy = // 执行原DTO的深拷贝逻辑
    return MessageBuilder.withPayload(copy).copyHeaders(message.getHeaders()).build();
}).handle(feedbackSubFlow()))

替代方案：PublishSubscribeChannel（多子流程并行场景） #

如果需要同时触发多个子流程监听同一步消息，可使用发布订阅通道。主流程将消息发送到发布订阅通道，主流程本身和各子流程作为独立订阅者，各自处理消息：

@Bean
public MessageChannel publishSubscribeChannel() {
    return MessageChannels.publishSubscribe().get();
}

@Bean
public IntegrationFlow mainFlow() {
    return IntegrationFlows.from(gateway())
            .transform(xmlToDtoTransformer())
            .filter(...)
            .channel(publishSubscribeChannel()) // 发送到发布订阅通道（触发反馈子流程）
            .transform(dtoSerializer())
            .handle(hmacSignHandler())
            .channel(publishSubscribeChannel()) // 再次发送（触发超时子流程）
            .channel("mainChannel")
            .get();
}

// 反馈子流程作为订阅者
@Bean
public IntegrationFlow feedbackSubFlow() {
    return IntegrationFlows.from(publishSubscribeChannel())
            .filter(...)
            .transform(...)
            .channel("feedbackChannel")
            .get();
}

// 超时记录子流程作为订阅者
@Bean
public IntegrationFlow timeoutRecordSubFlow() {
    return IntegrationFlows.from(publishSubscribeChannel())
            .transform(...)
            .handle(...)
            .get();
}

这种方式适合多子流程监听同一步消息的场景，但配置稍繁琐，优先推荐Wire Tap。

你之前用Wire Tap失败的可能原因 #

大概率是子流程误操作了原消息的引用（比如可变DTO被修改），或者错误配置了会影响主流程的组件。Wire Tap的消息是独立副本，子流程的任何处理都不会回溯到主流程，检查子流程是否存在共享状态或未做拷贝的可变对象即可解决。

内容的提问来源于stack exchange，提问作者usr-local-ΕΨΗΕΛΩΝ