一、Spring Integration 簡(jiǎn)介

Spring Integration 是 Spring 框架的擴(kuò)展，支持企業(yè)集成模式（EIP），提供輕量級(jí)的消息處理功能，幫助開(kāi)發(fā)者構(gòu)建可維護(hù)、可測(cè)試的企業(yè)集成解決方案。

核心目標(biāo)：

1. 提供簡(jiǎn)單的模型來(lái)實(shí)現(xiàn)復(fù)雜的企業(yè)集成。
2. 支持與外部系統(tǒng)的集成。
3. 提供模塊化、松耦合的消息處理架構(gòu)。

---

二、Spring Integration 核心組件

1. 消息（Message）

• 定義：消息是 Spring Integration 的核心，包含 payload（負(fù)載）和 header（頭部）。
• 創(chuàng)建消息：通過(guò) MessageBuilder 創(chuàng)建消息。

代碼示例：

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;Message<String> message = MessageBuilder.withPayload("Message Payload").setHeader("Message_Header1", "Header1_Value").setHeader("Message_Header2", "Header2_Value").build();

---

2. 消息通道（Message Channel）

• 定義：消息通道是消息傳遞的管道，連接消息的生產(chǎn)者和消費(fèi)者。
• 類(lèi)型：
  • 點(diǎn)對(duì)點(diǎn)（Point-to-Point）：每條消息最多被一個(gè)消費(fèi)者接收。
  • 發(fā)布/訂閱（Publish/Subscribe）：每條消息可以被多個(gè)訂閱者接收。
• 常見(jiàn)實(shí)現(xiàn)：
  • DirectChannel：默認(rèn)點(diǎn)對(duì)點(diǎn)通道。
  • NullChannel：虛擬通道，用于測(cè)試和調(diào)試。
  • 其他：PublishSubscribeChannel、QueueChannel、PriorityChannel 等。

---

3. 消息端點(diǎn)（Message Endpoint）

消息端點(diǎn)是應(yīng)用程序代碼與消息基礎(chǔ)設(shè)施之間的橋梁，主要類(lèi)型包括：

• Transformer：轉(zhuǎn)換消息內(nèi)容或結(jié)構(gòu)。
• Filter：過(guò)濾不符合條件的消息。
• Router：根據(jù)條件將消息路由到不同的通道。
• Splitter：將消息拆分為多個(gè)子消息。
• Aggregator：將多個(gè)消息聚合為一個(gè)消息。
• Service Activator：連接服務(wù)實(shí)例到消息系統(tǒng)。
• Channel Adapter：連接消息通道與外部系統(tǒng)。

---

三、貨物處理系統(tǒng)示例

1. 需求

實(shí)現(xiàn)一個(gè)貨物處理系統(tǒng)，功能包括：

1. 接收貨物消息。
2. 拆分貨物列表為單個(gè)貨物消息。
3. 基于重量過(guò)濾貨物。
4. 根據(jù)運(yùn)輸類(lèi)型（國(guó)內(nèi)/國(guó)際）路由貨物。
5. 轉(zhuǎn)換貨物消息。
6. 最終處理并記錄貨物信息。

---

2. 項(xiàng)目環(huán)境

• JDK：1.8
• Spring：4.1.2
• Spring Integration：4.1.0
• Maven：3.2.2
• 操作系統(tǒng)：Ubuntu 14.04

---

3. 完整代碼實(shí)現(xiàn)

Step 1：添加依賴(lài)

在 pom.xml 中添加 Spring 和 Spring Integration 的依賴(lài)：

<properties><spring.version>4.1.2.RELEASE</spring.version><spring.integration.version>4.1.0.RELEASE</spring.integration.version>
</properties><dependencies><!-- Spring 核心依賴(lài) --><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>${spring.version}</version></dependency><!-- Spring Integration 核心依賴(lài) --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-core</artifactId><version>${spring.integration.version}</version></dependency>
</dependencies>

---

Step 2：配置類(lèi)

創(chuàng)建 AppConfiguration 類(lèi)，配置消息通道和啟用 Spring Integration：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;@Configuration
@ComponentScan("com.onlinetechvision.integration")
@EnableIntegration
@IntegrationComponentScan("com.onlinetechvision.integration")
public class AppConfiguration {@Beanpublic MessageChannel cargoGWDefaultRequestChannel() {return new DirectChannel();}@Beanpublic MessageChannel cargoSplitterOutputChannel() {return new DirectChannel();}@Beanpublic MessageChannel cargoFilterOutputChannel() {return new DirectChannel();}@Beanpublic MessageChannel cargoTransformerOutputChannel() {return new DirectChannel();}
}

---

Step 3：消息網(wǎng)關(guān)

定義 CargoGateway 接口，作為消息系統(tǒng)的入口：

import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.messaging.Message;import java.util.List;@MessagingGateway
public interface CargoGateway {@Gateway(requestChannel = "cargoGWDefaultRequestChannel")void processCargoRequest(Message<List<Cargo>> message);
}

---

Step 4：消息拆分器

實(shí)現(xiàn) CargoSplitter，將貨物列表拆分為單個(gè)貨物消息：

import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Splitter;
import org.springframework.messaging.Message;import java.util.List;@MessageEndpoint
public class CargoSplitter {@Splitter(inputChannel = "cargoGWDefaultRequestChannel", outputChannel = "cargoSplitterOutputChannel")public List<Cargo> splitCargoList(Message<List<Cargo>> message) {return message.getPayload();}
}

---

Step 5：消息過(guò)濾器

實(shí)現(xiàn) CargoFilter，過(guò)濾重量超過(guò)限制的貨物：

import org.springframework.integration.annotation.Filter;
import org.springframework.integration.annotation.MessageEndpoint;@MessageEndpoint
public class CargoFilter {private static final double CARGO_WEIGHT_LIMIT = 1000.0;@Filter(inputChannel = "cargoSplitterOutputChannel", outputChannel = "cargoFilterOutputChannel", discardChannel = "cargoFilterDiscardChannel")public boolean filterCargo(Cargo cargo) {return cargo.getWeight() <= CARGO_WEIGHT_LIMIT;}
}

---

Step 6：服務(wù)激活器

實(shí)現(xiàn) CargoServiceActivator，處理最終的貨物消息：

import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.handler.annotation.Header;@MessageEndpoint
public class CargoServiceActivator {@ServiceActivator(inputChannel = "cargoTransformerOutputChannel")public void processCargo(Cargo cargo, @Header("CARGO_BATCH_ID") long batchId) {System.out.println("Processed Cargo: " + cargo + " in Batch: " + batchId);}
}

---

Step 7：運(yùn)行主程序

創(chuàng)建 Application 類(lèi)，初始化 Spring 容器并發(fā)送貨物請(qǐng)求：

import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.messaging.support.MessageBuilder;import java.util.Arrays;
import java.util.List;public class Application {public static void main(String[] args) {ApplicationContext context = new AnnotationConfigApplicationContext(AppConfiguration.class);CargoGateway gateway = context.getBean(CargoGateway.class);List<Cargo> cargos = Arrays.asList(new Cargo(1, "Receiver1", "Address1", 500, "Domestic"),new Cargo(2, "Receiver2", "Address2", 1500, "International"));gateway.processCargoRequest(MessageBuilder.withPayload(cargos).build());}
}

---

四、運(yùn)行過(guò)程

1. 啟動(dòng) Application 類(lèi)。
2. 系統(tǒng)會(huì)根據(jù)配置：
   • 拆分貨物列表。
   • 過(guò)濾重量超過(guò)限制的貨物。
   • 路由貨物到不同的通道。
   • 最終處理并記錄貨物信息。
3. 控制臺(tái)輸出處理結(jié)果。

---

五、適用場(chǎng)景

Spring Integration 非常適合以下場(chǎng)景：

1. 企業(yè)系統(tǒng)集成：如 ERP、CRM、供應(yīng)鏈系統(tǒng)之間的數(shù)據(jù)交換。
2. 消息驅(qū)動(dòng)架構(gòu)：如基于事件的微服務(wù)通信。
3. 復(fù)雜消息處理：如批量處理、過(guò)濾、路由、轉(zhuǎn)換等。
4. 與外部系統(tǒng)交互：如文件系統(tǒng)、消息隊(duì)列（RabbitMQ、Kafka）、數(shù)據(jù)庫(kù)等。

通過(guò) Spring Integration，可以輕松實(shí)現(xiàn)復(fù)雜的企業(yè)集成需求，同時(shí)保持代碼的可維護(hù)性和擴(kuò)展性。
參考鏈接：https://dzone.com/articles/message-processing-spring