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 框架開發

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

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

Spring Integration 基礎概念

起源

基本概念

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

區別

聯系

消息通道與消息端點

消息通道與消息端點

定義和配置消息通道

消息端點的作用和類型

消息處理器與適配器

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

消息處理器的使用方法

適配器與外部系統集成

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

消息的格式轉換與處理

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

集成模式與設計模式

Spring Integration 中常見的集成模式

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

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

實戰

1.添加依賴：

2.啟動類Application：

3.配置消息通道

4.Controller

5.訂單服務

6.支付處理服務

7.庫存檢查服務

8.發貨調度服務

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

10.測試

11.測試日志