spring-integration v6.4.9¶

为什么要使用 Spring Integration¶

想象一下，你正站在一个现代化的数据十字路口中央。左边，你的订单系统正以每秒数百次的速度喷涌着交易消息；右边，库存数据库沉默地更新着数字；前方，物流 API 和支付网关闪烁着等待调用的信号灯；身后，分析大屏饥渴地吞噬着每一条业务流水。而你，开发者，被期待成为那个优雅的交通指挥家，确保每一条信息准时、准确、安全地抵达目的地，还要在车流（数据流）激增时优雅地扩容，在某条道路（服务）暂时封闭时自动绕行。

这就是企业集成的真实战场——一个由异构系统、纷杂协议和脆弱连接构成的“巴别塔”。手动编写粘合代码？那很快会演变为一场维护噩梦。消息可能丢失，服务耦合会像藤蔓一样缠绕，任何一个环节的变动都将引发一场全局性的排查海啸。矛盾的核心在于：业务渴望敏捷与连贯，而技术栈的多样性与复杂性却不断制造着割裂与延迟。

此时，Spring Integration 如同一位经验丰富的城市规划师入场。它不承诺拆毁重建，而是提供一套经过验证的“交通模式”（企业集成模式）和现成的“交通设施”（连接器、通道、端点）。使用它，意味着你选择用声明式的蓝图（配置）替代硬编码的指挥手势，用内建的容错机制替代手动的重试循环，用清晰的消息流管道替代隐藏在业务逻辑中的跳跃调用。它的价值，在于将你从“通信泥潭”的战术挣扎中解放出来，让你能专注于构建业务价值本身这项战略任务。 你不是在编写又一个集成脚本，你是在描绘一幅可持续、可观测、可扩展的数据流动景观。

Spring Integration 是什么¶

简而言之，Spring Integration 是“企业集成模式”在 Spring 生态中的具象化实现。它扩展了 Spring 编程模型，为核心哲学——“约定优于配置”——赋予了处理消息流的能力。

你可以将它理解为一个高度专业化的 Spring 框架子集。它提供了统一的编程模型，通过“消息（Message）”这个核心抽象，来连接系统内外部的各个组件。框架内置了大量的“适配器（Adapter）”和“通道（Channel）”，让你能够用极少的代码，将诸如 HTTP 请求、文件变动、数据库记录、消息队列（RabbitMQ, Kafka）、TCP/UDP 包等事件，转化为统一的消息，并在预先定义好的“管道”中路由、转换、过滤和处理，最终驱动业务逻辑的执行。

它的本质是一个**轻量级消息驱动中间件**，但其魅力在于深度融入 Spring 生态，让你能以熟悉的方式（如注解、Java DSL、XML配置）来构建复杂、健壮的集成流。

入门示例¶

真实场景：电商订单的异步处理与状态更新

假设我们有一个简易的电商系统。当用户下单后，我们需要： 1. 持久化订单到数据库。 2. 异步调用库存服务，扣减库存。 3. 根据库存结果，更新订单状态。 4. 将处理完成的订单事件发布到消息队列，供其他系统（如物流、分析）消费。

不使用 Spring Integration，你可能会在订单服务中手动创建线程池、编写 REST 客户端调用库存服务、处理回调、再手动发送消息到 Kafka。代码很快会变得冗长且难以维护。

使用 Spring Integration，我们可以优雅地构建一个消息流：

@Configuration
@EnableIntegration
public class OrderProcessingConfig {

    // 1. 定义一个接收订单的网关（入口）
    @MessagingGateway
    public interface OrderGateway {
        @Gateway(requestChannel = "order.input")
        void placeOrder(Order order);
    }

    // 2. 定义集成流
    @Bean
    public IntegrationFlow orderProcessingFlow() {
        return IntegrationFlow.from("order.input")
                // 3. 持久化订单（服务激活）
                .handle(orderService, "save")
                // 4. 异步调用库存服务（通道间引入执行器）
                .channel(MessageChannels.executor(taskExecutor()))
                .handle(InventoryService.class, "deduct")
                // 5. 根据库存结果路由：成功->确认，失败->取消
                .<InventoryResponse, Boolean>route(
                        InventoryResponse::isSuccess,
                        mapping -> mapping
                                .subFlowMapping(true, sf -> sf.handle(orderService, "confirm"))
                                .subFlowMapping(false, sf -> sf.handle(orderService, "cancel"))
                )
                // 6. 最终事件发布到 Kafka
                .handle(kafkaTemplate::send)
                .get();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        return new ThreadPoolTaskExecutor();
    }
}

发生了什么？ * OrderGateway 是一个普通接口，但被 @MessagingGateway 修饰后，框架会为其生成实现。调用 placeOrder() 即向消息流发送一个订单消息。 * 整个流程通过流畅的 Java DSL 定义，清晰可视。 * .channel(MessageChannels.executor(...)) 轻松引入了异步处理，无需手动管理线程。 * .route() 操作符根据库存结果进行条件路由，逻辑一目了然。 * 每个 .handle() 都连接着一个 Spring Bean 的业务方法，保持了业务代码的纯粹性。

这个流是声明式的、可测试的、且易于扩展。如果需要添加新的处理步骤（如风控检查），只需在流中插入一个 .handle() 即可。这就是 Spring Integration 带来的秩序与简洁。

Spring Integration v6.4.9版本更新了什么¶

经查阅项目发布页，用户所指应为 v6.4.0 版本。这是一个次要版本更新，核心围绕 Spring 框架 6.2 和 Spring Boot 3.4 的兼容性展开。主要更新可概括为： 1. 版本对齐：项目升级至与 Spring Framework 6.2.10 和 Spring Boot 3.4.0 兼容，确保了在新一代 Spring 生态中的稳定运行。 2. 依赖升级：同步更新了多个核心依赖项，如 Reactor 2024.0.0、Jackson 2.18.2 等，以获得性能改进和安全性修复。 3. 新特性引入：提供了对 Apache Pulsar 客户端的新版本支持，扩展了消息处理能力。 4. 问题修复：包含了一系列针对前期版本中发现问题的修复，提升了框架整体的稳定性和可靠性。 5. 维护性改进：这是一个持续维护和优化的常规版本，旨在为用户提供更平滑的升级路径和更坚实的基础。

更新日志¶

🐞 错误修复¶

修复了 CompositeExecutor 构造函数中不正确的空值检查。
解决了 Redis[Clustered] 在解锁时出现 “CROSSSLOT Keys in request” 错误的问题。
修复了应用程序关闭期间，DefaultHeaderChannelRegistry 中存在的竞态条件导致错误日志的问题。

🔨 依赖升级¶

将 `org.springframework.kafka:spring-kafka-b