Spring Integration 輕松實現服務間消息傳遞，真香！

作者：不才陳某 2025-04-09 02:02:00

開發前端

Spring Integration 是 Spring 框架的一個重要擴展，其核心目標在于極大地簡化企業集成模式的開發過程。它構建了一種基于消息的編程模型，讓分布式系統中的系統集成變得更加輕松便捷。

在當今分布式系統的背景下，如何優雅地實現系統之間的消息傳遞是每個開發者都關心的話題。

而Spring Integration，作為Spring家族的一員，正是為了解決這個難題而生。

在這篇文章中，我們將踏上穿越消息之路，深入探討Spring Integration的魅力。

Spring Integration 基礎概念

起源

基本概念

消息：在 Spring Integration 的體系中，消息是信息傳遞的關鍵載體。它就像一個裝滿各種信息的“包裹”，不僅可以包含業務數據，還能攜帶頭部信息、消息標簽等內容。消息會沿著特定的通道（Channel）在系統中有序傳遞。
通道（Channel）：通道就像是消息在系統中流動的“高速公路”。Spring Integration 提供了多種不同類型的通道，例如直接通道（Direct Channel），它就像一條直達專線，能讓消息快速高效地傳遞；發布 - 訂閱通道（Publish - Subscribe Channel），類似于廣播電臺，可以將消息同時傳遞給多個訂閱者；隊列通道（Queue Channel），如同排隊等待服務的隊伍，消息會按照順序依次進行處理。關注公眾號：碼猿技術專欄，回復關鍵詞：1111 獲取阿里內部Java性能調優手冊！
端點（Endpoint）：端點是消息的生產者或者消費者，它們就像接力賽中的運動員，消息從一個端點傳遞到另一個端點，從而形成一個完整的消息處理流程。
適配器（Adapter）：適配器是 Spring Integration 與外部系統或者服務之間的“橋梁”。它能夠將外部系統的消息“翻譯”成 Spring Integration 能夠理解的消息格式，也可以將 Spring Integration 的消息傳遞給外部系統。
過濾器（Filter）：過濾器就像是一個嚴格的“門衛”，只有滿足特定條件的消息才能通過它的“檢查”。它在消息的路由、轉換等過程中發揮著重要作用。
轉換器（Transformer）：轉換器如同一個神奇的“魔法師”，能夠將消息從一種形式轉換為另一種形式，以滿足系統的不同需求。它可以對數據格式進行轉換，也可以修改消息體的內容。

Spring Integration 與傳統消息中間件的區別與聯系

區別

Spring Integration 是框架：Spring Integration 是基于 Spring 構建的一個強大框架，它提供了一整套用于構建企業集成模式的工具和組件，就像一個功能齊全的“工具箱”。
傳統消息中間件是產品：傳統消息中間件通常是獨立的產品，如 RabbitMQ、Apache Kafka、ActiveMQ 等，它們專注于提供可靠的消息傳遞服務，就像專業的“快遞物流公司”。

聯系

整合性：Spring Integration 具有強大的整合能力，它可以與傳統消息中間件完美集成。通過適配器，Spring Integration 能夠與外部消息中間件進行通信，就像一個萬能的“接口”，幫助企業集成系統與不同的消息中間件進行對接。
解耦與異步通信：和傳統消息中間件一樣，Spring Integration 也支持解耦和異步通信的模式。通過消息的發布與訂閱，系統組件之間可以實現解耦和松耦合，就像各個部門之間通過郵件進行溝通，彼此獨立又能協同工作。
消息傳遞：Spring Integration 和傳統消息中間件都基于消息傳遞的模型。消息作為信息的載體，在系統中傳遞，實現不同組件之間的通信，就像信件在不同的收件人之間傳遞一樣。

總體而言，Spring Integration 提供了一種更加輕量級和靈活的方式來實現企業集成，而傳統消息中間件更專注于提供可靠的消息傳遞服務。在實際應用中，我們可以根據具體的需求選擇合適的技術和工具。

消息通道與消息端點

定義和配置消息通道

定義消息通道：在 Spring Integration 中，消息通道是消息在系統中傳遞的關鍵“管道”。我們可以使用 XML 配置或者 Java 代碼來定義消息通道。

XML 配置示例：

<int:channel id="myChannel"/>

Java 配置示例：

@Bean
public MessageChannel myChannel() {
    return MessageChannels.direct().get();
}

配置消息通道的類型：Spring Integration 提供了多種不同類型的消息通道，如直接通道（Direct Channel）、發布 - 訂閱通道（Publish - Subscribe Channel）、隊列通道（Queue Channel）等。我們可以根據實際需求選擇合適的通道類型。

XML 配置示例：

<!-- 配置直接通道 -->
<int:channel id="directChannel"/>
<!-- 配置發布 - 訂閱通道 -->
<int:publish-subscribe-channel id="publishSubscribeChannel"/>
<!-- 配置隊列通道 -->
<int:queue-channel id="queueChannel"/>

Java 配置示例：

消息通道的屬性配置：我們還可以通過配置消息通道的一些屬性，如容量、過期時間等，來滿足具體的需求。

XML 配置示例：

<int:channel id="myChannel" capacity="10" />

Java 配置示例：

@Bean
public MessageChannel myChannel() {
    return MessageChannels.direct().capacity(10).get();
}

消息端點的作用和類型

作用：消息端點是消息的生產者或者消費者，它定義了消息的處理邏輯。消息從一個端點流向另一個端點，形成一個完整的消息處理流程。
消息端點的類型：

過濾器（Filter）：用于過濾消息，只有滿足特定條件的消息才能通過。它就像一個“篩子”，篩選出符合要求的消息。
轉換器（Transformer）：用于將消息從一種形式轉換為另一種形式。它就像一個“變形金剛”，將消息變成不同的形態。

分發器（Dispatcher）：用于將消息分發給不同的子通道，根據條件進行消息路由。它就像一個“交通指揮員”，根據不同的規則將消息引導到不同的方向。
服務激活器（Service Activator）：用于將消息傳遞給特定的服務進行處理。它就像一個“調度員”，將消息分配給合適的服務進行處理。
消息處理器（Message Handler）：用于處理消息，可以是一個 Java 方法、表達式、腳本等。它就像一個“工人”，負責對消息進行具體的處理。
消息源（Message Source）：用于產生消息的端點，例如文件輸入、JDBC 查詢等。它就像消息的“源頭”，不斷地產生新的消息。
通道適配器（Channel Adapter）：用于將外部系統的消息轉換為 Spring Integration 的消息格式。它就像一個“翻譯官”，幫助不同系統之間進行消息的“交流”。---》企業級實戰總結40講
消息生產者端點：
消息消費者端點：
消息路由器端點：
其他類型：
配置消息端點：消息端點可以通過 XML 配置或者 Java 代碼進行定義。
XML 配置示例：

<int:service-activator input-channel="myChannel" ref="myService" method="processMessage"/>

Java 配置示例：

@ServiceActivator(inputChannel = "myChannel")
public void processMessage(Message<String> message) {
    // 處理消息的邏輯
}

通過合理定義和配置消息通道以及消息端點，我們可以構建出靈活、可擴展的消息傳遞系統，實現消息在系統中的高效流動和處理。關注公眾號：碼猿技術專欄，回復關鍵詞：1111 獲取阿里內部Java性能調優手冊！

消息處理器與適配器

消息處理器與適配器在 Spring Integration 中的使用

消息處理器的使用方法

消息處理器是 Spring Integration 中用于處理消息的核心組件，它可以是一個 Java 方法、表達式、腳本等。以下是消息處理器的使用方法：

Java 方法處理器：

@ServiceActivator(inputChannel = "inputChannel")
public void handleMessage(String message) {
    // 處理消息的邏輯
    System.out.println("Received Message: " + message);
}

在上述代碼中，handleMessage 方法是一個消息處理器，通過 @ServiceActivator 注解將其與名為 inputChannel 的輸入通道關聯起來。當消息被發送到該通道時，該方法會被調用來處理消息。

表達式處理器：

<int:service-activator input-channel="inputChannel" expression="@myService.process(#payload)">
    <int:poller fixed-rate="1000"/>
</int:service-activator>

在上述配置中，expression 屬性定義了一個表達式，指定了消息處理的邏輯。這個表達式將調用名為 process 的方法，#payload 表示消息的載荷。

適配器與外部系統集成

適配器用于將外部系統的消息與 Spring Integration 進行集成，使得外部系統的消息能夠在 Spring Integration 中流通。以下是適配器的使用方法：

文件適配器：

<int-file:inbound-channel-adapter id="filesIn"
                                 channel="inputChannel"
                                 directory="file:${java.io.tmpdir}/input">
    <int:poller fixed-rate="5000"/>
</int-file:inbound-channel-adapter>

上述配置使用文件適配器（<int-file:inbound-channel-adapter>）來監聽指定目錄中的文件，并將文件內容發送到名為 inputChannel 的通道。

JDBC 適配器：

<int-jdbc:inbound-channel-adapter id="jdbcInboundAdapter"
                                  query="SELECT * FROM my_table"
                                  channel="inputChannel">
    <int:poller fixed-rate="10000"/>
</int-jdbc:inbound-channel-adapter>

上述配置中，JDBC 適配器（<int-jdbc:inbound-channel-adapter>）從數據庫執行查詢，并將結果發送到 inputChannel 通道。

HTTP 適配器：

<int-http:inbound-channel-adapter id="httpInboundAdapter"
                                  channel="inputChannel"
                                  path="/receiveMessage"
                                  request-mapper="requestMapping">
    <int:poller fixed-rate="10000"/>
</int-http:inbound-channel-adapter>

上述配置使用 HTTP 適配器（<int-http:inbound-channel-adapter>）監聽指定路徑的 HTTP 請求，并將請求的消息發送到 inputChannel 通道。

以上示例展示了如何使用不同類型的適配器來與外部系統進行集成。適配器將外部系統的消息轉換為 Spring Integration 的消息，并通過通道在整個系統中傳遞。適配器的配置取決于具體的集成需求和外部系統的特性。---》企業級實戰總結40講

消息轉換與路由在 Spring Integration 中的應用

消息的格式轉換與處理

消息轉換是 Spring Integration 中常見的操作，用于將消息從一種格式或結構轉換為另一種格式或結構，以滿足系統的需求。以下是消息轉換的實際應用場景和示例：

JSON 到對象的轉換：

@Transformer(inputChannel = "jsonInputChannel", outputChannel = "objectOutputChannel")
public MyObject convertJsonToObject(String jsonString) {
    // 使用 Jackson 庫將 JSON 字符串轉換為 Java 對象
    return objectMapper.readValue(jsonString, MyObject.class);
}

在上述代碼中，@Transformer 注解表示這是一個消息轉換器，它將 jsonInputChannel 通道的 JSON 消息轉換為 Java 對象，并將結果發送到 objectOutputChannel 通道。

對象到 JSON 的轉換：

@Transformer(inputChannel = "objectInputChannel", outputChannel = "jsonOutputChannel")
public String convertObjectToJson(MyObject myObject) {
    // 使用 Jackson 庫將 Java 對象轉換為 JSON 字符串
    return objectMapper.writeValueAsString(myObject);
}

在這個例子中，消息轉換器將 objectInputChannel 通道的 Java 對象轉換為 JSON 字符串，并將結果發送到 jsonOutputChannel 通道。

路由器的作用和實際應用場景

路由器用于根據消息的內容或特征將消息路由到不同的通道，實現消息在系統中的分發。以下是路由器的實際應用場景和示例：

內容路由器：

<int:router input-channel="inputChannel" expression="payload.type">
    <int:mapping value="A" channel="channelA"/>
    <int:mapping value="B" channel="channelB"/>
    <int:mapping value="C" channel="channelC"/>
</int:router>

在上述配置中，內容路由器（<int:router>）根據消息的 type 屬性的值將消息路由到不同的通道。如果消息的 type 是 "A"，則路由到 channelA；如果是 "B"，則路由到 channelB，以此類推。

篩選器路由器：

<int:router input-channel="inputChannel">
    <int:mapping value="payload.type == 'A'" channel="channelA"/>
    <int:mapping value="payload.type == 'B'" channel="channelB"/>
    <int:mapping value="payload.type == 'C'" channel="channelC"/>
</int:router>

在這個例子中，路由器根據篩選條件將消息路由到不同的通道。只有滿足條件的消息才會被路由到相應的通道。

路由器的靈活性使得我們可以根據消息的內容、屬性或條件進行動態的路由，從而實現系統中不同組件的消息處理邏輯的分離。路由器的配置可以根據具體的需求進行調整，以適應不同的應用場景。

集成模式與設計模式

Spring Integration 中常見的集成模式

Spring Integration 提供了許多常見的集成模式，這些模式能夠幫助開發人員構建可靠、可擴展的消息驅動系統。以下是一些常見的集成模式：

消息通道（Message Channel）：它定義了消息在系統中傳遞的路徑，是消息傳遞的重要媒介，就像城市中的道路，消息沿著它在系統中流動。
消息端點（Message Endpoint）：定義了消息的生產者或者消費者，可以是服務激活器、消息處理器等。它就像道路上的車站，負責消息的發送和接收。
消息適配器（Message Adapter）：用于將外部系統的消息轉換為 Spring Integration 的消息格式，實現系統與外部系統的集成。它就像一個翻譯官，幫助不同語言的系統進行交流。
消息網關（Message Gateway）：提供了對系統的入口，允許外部系統通過網關發送消息到系統中，或者從系統中獲取消息。它就像系統的大門，控制著消息的進出。
消息轉換器（Message Transformer）：用于對消息的格式進行轉換，將消息從一種表示形式轉換為另一種，以滿足系統的需求。它就像一個變形金剛，能把消息變成不同的樣子。
消息過濾器（Message Filter）：用于過濾消息，只有滿足特定條件的消息才能通過，實現對消息的篩選。它就像一個篩子，把不符合要求的消息過濾掉。
消息路由器（Message Router）：根據消息的內容、屬性或條件將消息路由到不同的通道，實現消息的分發。它就像一個交通指揮員，根據不同的規則將消息引導到不同的方向。
聚合器（Aggregator）：將多個相關的消息合并為一個消息，通常用于處理分散的消息片段。它就像一個拼圖高手，把分散的消息碎片拼成完整的消息。
分裂器（Splitter）：將一個消息拆分為多個消息，通常用于處理大塊的消息內容。它就像一個切割工人，把大的消息切割成小塊。
定時器（Timer）：定期發送消息，用于實現定時任務或者輪詢外部系統。它就像一個鬧鐘，定時提醒系統執行相應的操作。

如何根據設計模式構建消息驅動的系統

在構建消息驅動的系統時，我們可以借鑒一些設計模式來提高系統的可維護性、可擴展性和可測試性。以下是一些常用的設計模式，特別是在消息驅動系統中的應用：

發布 - 訂閱模式（Publish - Subscribe Pattern）：在消息驅動系統中，通過使用發布 - 訂閱模式可以實現消息的廣播，允許多個組件訂閱并接收相同的消息。它就像一個廣播電臺，向多個聽眾同時發送消息。
觀察者模式（Observer Pattern）：觀察者模式可以用于實現消息的訂閱和通知機制，在消息產生時通知所有的觀察者。它就像一個新聞發布系統，當有新聞發布時，會通知所有訂閱的用戶。
策略模式（Strategy Pattern）：策略模式可用于實現靈活的消息處理策略，根據不同的需求選擇不同的消息處理算法。它就像一個工具箱，根據不同的任務選擇不同的工具。
裝飾者模式（Decorator Pattern）：裝飾者模式可用于動態地添加消息處理邏輯，如消息轉換器、消息過濾器等。它就像給消息穿上不同的衣服，增加不同的功能。
責任鏈模式（Chain of Responsibility Pattern）：責任鏈模式可用于實現消息處理管道，每個處理器負責處理特定類型的消息，形成一個處理鏈。它就像一個流水線，每個工人負責完成特定的工序。
.命令模式（Command Pattern）：命令模式可以將消息封裝為命令對象，以支持撤銷、重做等操作。
工廠模式（Factory Pattern）：工廠模式可用于創建消息適配器、消息處理器等組件，提供一種靈活的對象創建方式。

Spring Integration中流程和通道攔截的實現方法

在Spring Integration中，可以通過攔截器（Interceptor）來對消息通道和流程進行攔截和處理。攔截器允許在消息在通道中傳遞和處理的過程中執行自定義邏輯。

1. 通道攔截：

在通道級別，可以使用通道攔截器來對消息通道的發送和接收進行攔截。

<int:channel id="myChannel">
    <int:interceptors>
        <int:wire-tap channel="logChannel"/>
    </int:interceptors>
</int:channel>

上述配置中，<int:wire-tap>是一個通道攔截器，將通道上的所有消息發送到logChannel通道，以便記錄日志或進行其他操作。

2. 流程攔截：

在流程級別，可以使用<int:advice>和<int:expression-advice>等元素來添加攔截器。

<int:service-activator input-channel="inputChannel" output-channel="outputChannel">
    <int:advice-chain>
        <int:expression-advice expression="payload.toUpperCase()"/>
    </int:advice-chain>
</int:service-activator>

在上述配置中，<int:expression-advice>是一個流程攔截器，它使用SpEL表達式將消息內容轉換為大寫。

攔截器的應用和自定義：

1. 內置攔截器的應用：

Spring Integration提供了一些內置的攔截器，如WireTap、LoggingHandler等，用于實現常見的攔截需求。例如：

<int:channel id="inputChannel">
    <int:interceptors>
        <int:wire-tap channel="logChannel"/>
    </int:interceptors>
</int:channel>

上述配置中，使用了內置的WireTap攔截器，將通道上的所有消息發送到logChannel通道。

2. 自定義攔截器：

可以通過實現ChannelInterceptor接口或擴展ChannelInterceptorAdapter類來創建自定義的通道攔截器。同樣，通過實現Advice接口或擴展AbstractRequestHandlerAdvice類可以創建自定義的流程攔截器。

<int:service-activator input-channel="inputChannel" output-channel="outputChannel">
    <int:advice-chain>
        <bean class="com.example.CustomExpressionAdvice"/>
    </int:advice-chain>
</int:service-activator>

上述配置中，使用了自定義的流程攔截器CustomExpressionAdvice，該類需實現Advice接口。

通過應用內置或自定義的攔截器，可以在消息處理的不同階段執行自定義的邏輯，如日志記錄、性能監控、消息轉換等。

實戰

傳統訂單處理流程往往涉及多個手動步驟，容易導致延遲和錯誤。為了提高電商平臺的運作效率，客戶那邊要求我們開發一個自動化訂單處理系統，從訂單創建到支付、庫存檢查和發貨全流程自動化處理，通過消息觸發相關的業務邏輯，減少人為失誤。

1.添加依賴：

2.啟動類Application：

3.配置消息通道

/**
 * 配置消息通道
 */
@Configuration
publicclass IntegrationConfig {

    /**
     * 定義訂單創建的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel orderCreatedChannel() {
        returnnew DirectChannel();
    }

    /**
     * 定義支付處理的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel paymentProcessedChannel() {
        returnnew DirectChannel();
    }

    /**
     * 定義庫存檢查的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel inventoryCheckedChannel() {
        returnnew DirectChannel();
    }

    /**
     * 定義發貨調度的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel shipmentScheduledChannel() {
        returnnew DirectChannel();
    }
}

4.Controller

5.訂單服務

6.支付處理服務

7.庫存檢查服務

8.發貨調度服務

9.訂單處理相關的消息網關接口

10.測試

curl -X POST http://localhost:8080/orders \
     -H "Content-Type: application/json" \
     -d '{"orderId": "123", "productId": "P001", "quantity": 2}'

11.測試日志

Creating order: 123
Handling order creation for: 123
Processing payment for order: 123
Checking inventory for product: P001
Product is in stock.
Scheduling shipment for order: 123
Shipment scheduled for order: 123

責任編輯：武曉燕來源：碼猿技術專欄

Spring 框架開發