一篇非常好的Spring Integration 教程
更新時(shí)間：2026年06月15日 10:19:11 作者：Full Stack Developme
Spring Integration 是 Spring 家族里專門用來(lái)搞定?企業(yè)應(yīng)用集成?的框架,核心就是讓不同系統(tǒng)之間能?松耦合?地通過(guò)?消息?來(lái)通信,本文給大家介紹Spring Integration 教程,感興趣的朋友一起看看吧
一、什么是 Spring Integration
Spring Integration 是 Spring 生態(tài)系統(tǒng)中的一個(gè)擴(kuò)展模塊，用于實(shí)現(xiàn)企業(yè)應(yīng)用集成 (EAI, Enterprise Application Integration)。它基于 Spring 框架，提供了一套聲明式的適配器，用于集成不同的系統(tǒng)和服務(wù)。
核心特點(diǎn)：
- 基于消息驅(qū)動(dòng)的架構(gòu)
- 支持多種傳輸協(xié)議（HTTP, TCP, JMS, AMQP, FTP, File 等）
- 提供開(kāi)箱即用的端點(diǎn)適配器
- 支持企業(yè)集成模式 (EIP, Enterprise Integration Patterns)
二、核心概念
1. Message
// 消息由消息頭和消息體組成
public interface Message<T> {
T getPayload();
MessageHeaders getHeaders();
}
// 創(chuàng)建消息
Message<String> message = MessageBuilder.withPayload("Hello")
.setHeader("key", "value")
.build();2. Message Channel
消息通道用于在發(fā)送者和接收者之間傳遞消息。
// 點(diǎn)對(duì)點(diǎn)通道
@Bean
public MessageChannel directChannel() {
return new DirectChannel();
}
// 發(fā)布訂閱通道
@Bean
public MessageChannel publishSubscribeChannel() {
return new PublishSubscribeChannel();
}
// 隊(duì)列通道
@Bean
public MessageChannel queueChannel() {
return new QueueChannel(10);
}3. Message Endpoint
消息端點(diǎn)負(fù)責(zé)處理消息。
三、快速入門示例
Maven 依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<!-- 可選：特定協(xié)議支持 -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-http</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
</dependency>基礎(chǔ)配置示例
@Configuration
@EnableIntegration
public class IntegrationConfig {
// 定義消息通道
@Bean
public MessageChannel inputChannel() {
return new DirectChannel();
}
@Bean
public MessageChannel outputChannel() {
return new DirectChannel();
}
// 定義集成流程
@Bean
public IntegrationFlow simpleFlow() {
return IntegrationFlow.from(inputChannel())
.transform(String.class, s -> s.toUpperCase())
.filter(s -> s.startsWith("A"))
.handle(System.out::println)
.get();
}
}使用 @MessagingGateway
// 定義網(wǎng)關(guān)接口
@MessagingGateway(defaultRequestChannel = "inputChannel")
public interface SimpleGateway {
void sendMessage(String message);
@Gateway(requestChannel = "requestChannel", replyChannel = "replyChannel")
String sendAndReceive(String message);
}
// 使用網(wǎng)關(guān)
@Service
public class MessageService {
@Autowired
private SimpleGateway gateway;
public void send(String message) {
gateway.sendMessage(message);
}
}四、常用企業(yè)集成模式
1. 消息轉(zhuǎn)換器 (Transformer)
@Bean
public IntegrationFlow transformerFlow() {
return IntegrationFlow.from("inputChannel")
.transform(new GenericTransformer<String, User>() {
@Override
public User transform(String source) {
return new User(source);
}
})
.channel("outputChannel")
.get();
}2. 消息過(guò)濾器 (Filter)
@Bean
public IntegrationFlow filterFlow() {
return IntegrationFlow.from("inputChannel")
.filter(payload -> payload instanceof User)
.filter("payload.age > 18") // SpEL 表達(dá)式
.channel("adultChannel")
.get();
}3. 消息路由器 (Router)
@Bean
public IntegrationFlow routerFlow() {
return IntegrationFlow.from("inputChannel")
.route(payload -> {
if (payload instanceof Order) return "orderChannel";
if (payload instanceof Payment) return "paymentChannel";
return "errorChannel";
})
.get();
}4. 消息拆分器 (Splitter) 和聚合器 (Aggregator)
@Bean
public IntegrationFlow splitterAggregatorFlow() {
return IntegrationFlow.from("inputChannel")
.split() // 拆分消息
.channel("splitChannel")
.aggregate() // 聚合消息
.channel("outputChannel")
.get();
}五、常用適配器示例
1. 文件適配器
@Configuration
public class FileIntegrationConfig {
// 讀取文件
@Bean
public IntegrationFlow fileReaderFlow() {
return IntegrationFlow.from(
Files.inboundAdapter(new File("/input"))
.patternFilter("*.txt"),
e -> e.poller(Pollers.fixedDelay(1000))
)
.transform(File.class, File::getAbsolutePath)
.handle(System.out::println)
.get();
}
// 寫入文件
@Bean
public IntegrationFlow fileWriterFlow() {
return IntegrationFlow.from("fileInputChannel")
.handle(Files.outboundAdapter(new File("/output"))
.autoCreateDirectory(true))
.get();
}
}2. HTTP 適配器
@Configuration
public class HttpIntegrationConfig {
// HTTP 入站網(wǎng)關(guān)
@Bean
public IntegrationFlow httpInboundFlow() {
return IntegrationFlow.from(
Http.inboundGateway("/api/message")
.requestMapping(m -> m.methods(HttpMethod.POST))
.requestPayloadType(String.class)
.replyTimeout(30000)
)
.transform(String.class, s -> "Processed: " + s)
.get();
}
// HTTP 出站網(wǎng)關(guān)
@Bean
public IntegrationFlow httpOutboundFlow() {
return IntegrationFlow.from("requestChannel")
.handle(Http.outboundGateway("https://api.example.com/data")
.httpMethod(HttpMethod.GET)
.expectedResponseType(String.class))
.channel("responseChannel")
.get();
}
}3. JMS 適配器
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jms</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-broker</artifactId>
</dependency>@Configuration
public class JmsIntegrationConfig {
@Bean
public ConnectionFactory connectionFactory() {
return new ActiveMQConnectionFactory("tcp://localhost:61616");
}
// JMS 入站適配器
@Bean
public IntegrationFlow jmsInboundFlow() {
return IntegrationFlow.from(
Jms.inboundAdapter(connectionFactory())
.destination("queue.in")
)
.transform(String.class, String::toUpperCase)
.handle(message -> System.out.println("Received: " + message))
.get();
}
// JMS 出站適配器
@Bean
public IntegrationFlow jmsOutboundFlow() {
return IntegrationFlow.from("jmsOutputChannel")
.handle(Jms.outboundAdapter(connectionFactory())
.destination("queue.out"))
.get();
}
}六、高級(jí)特性
1. 錯(cuò)誤處理
@Bean
public IntegrationFlow errorHandlingFlow() {
return IntegrationFlow.from("inputChannel")
.transform(...)
.handle(..., e -> e
.advice(ExpressionEvaluatingRequestHandlerAdvice.class)
.advice(advice -> advice
.onFailureExpression("payload.message")
.trapException(true))
)
.get();
}
// 全局錯(cuò)誤通道
@Bean
public IntegrationFlow errorFlow() {
return IntegrationFlow.from(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
.handle(message -> {
Exception exception = (Exception) message.getPayload();
log.error("Error: ", exception);
})
.get();
}2. 消息歷史
@Configuration
@EnableIntegration
@EnableMessageHistory
public class HistoryConfig {
@Bean
public IntegrationFlow historyFlow() {
return IntegrationFlow.from("inputChannel")
.transform(...)
.enrichHeaders(s -> s.header(MessageHistory.HEADER_NAME, new MessageHistory()))
.handle(...)
.get();
}
}3. 控制總線
@Bean
public IntegrationFlow controlBusFlow() {
return IntegrationFlow.from("controlBus")
.controlBus()
.get();
}
// 使用控制總線
@Component
public class ControlBusService {
@Autowired
@Qualifier("controlBus")
private MessageChannel controlBus;
public void stopChannel() {
controlBus.send(MessageBuilder.withPayload("@myChannel.stop()").build());
}
}七、完整示例：文件處理系統(tǒng)
@SpringBootApplication
@EnableIntegration
public class FileProcessingApplication {
public static void main(String[] args) {
SpringApplication.run(FileProcessingApplication.class, args);
}
}
@Configuration
public class FileProcessingFlow {
private static final Logger log = LoggerFactory.getLogger(FileProcessingFlow.class);
// 文件輸入目錄
@Value("${input.directory:/input}")
private String inputDirectory;
// 處理成功目錄
@Value("${success.directory:/success}")
private String successDirectory;
// 處理失敗目錄
@Value("${failed.directory:/failed}")
private String failedDirectory;
@Bean
public IntegrationFlow fileProcessingFlow() {
return IntegrationFlow.from(
Files.inboundAdapter(new File(inputDirectory))
.patternFilter("*.csv")
.preventDuplicates(true)
.autoCreateDirectory(true),
e -> e.poller(Pollers.fixedDelay(5000)
.maxMessagesPerPoll(5)
.advice(expressionAdvice()))
)
.channel(MessageChannels.queue("processingChannel", 10))
.transform(Files.toStringTransformer()) // 文件轉(zhuǎn)字符串
.split(s -> s.delimiters("\n")) // 按行拆分
.filter(line -> !line.trim().isEmpty())
.transform(line -> parseCsvLine(line)) // 解析CSV
.aggregate(aggregatorSpec -> aggregatorSpec
.releaseStrategy(new SimpleSequenceSizeReleaseStrategy())
.correlationStrategy(message -> "batch"))
.handle(message -> processBatch((List<Map<String, String>>) message.getPayload()))
.handle(Files.outboundAdapter(new File(successDirectory))
.autoCreateDirectory(true)
.fileNameGenerator(message -> generateFileName(message)))
.get();
}
// 錯(cuò)誤處理：失敗的文件移動(dòng)到失敗目錄
@Bean
public IntegrationFlow errorHandlingFlow() {
return IntegrationFlow.from(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
.handle(message -> {
Message<?> failedMessage = (Message<?>) message.getHeaders().get("inputMessage");
File failedFile = (File) failedMessage.getPayload();
FileUtils.moveFileToDirectory(failedFile, new File(failedDirectory), true);
log.error("Failed to process file: {}", failedFile.getName());
})
.get();
}
private Map<String, String> parseCsvLine(String line) {
// CSV解析邏輯
return new HashMap<>();
}
private void processBatch(List<Map<String, String>> batch) {
// 批量處理邏輯
log.info("Processing batch of {} records", batch.size());
}
private String generateFileName(Message<?> message) {
return "processed_" + System.currentTimeMillis() + ".json";
}
@Bean
public Advice expressionAdvice() {
return new ExpressionEvaluatingRequestHandlerAdvice();
}
}八、最佳實(shí)踐
- 合理使用通道類型：DirectChannel 用于同步，QueueChannel 用于緩沖，PublishSubscribeChannel 用于廣播
- 避免阻塞操作：使用 QueueChannel 時(shí)注意配置合適的大小和 poller
- 錯(cuò)誤處理：始終配置錯(cuò)誤通道，記錄異常并適當(dāng)重試
- 監(jiān)控和管理：使用 Spring Boot Actuator 監(jiān)控集成端點(diǎn)
management:
endpoints:
web:
exposure:
include: integration測(cè)試：使用 @SpringIntegrationTest 進(jìn)行集成測(cè)試
@SpringBootTest
@SpringIntegrationTest(noAutoStartup = {"inputChannel"})
class IntegrationFlowTest {
@Test
void testFlow() {
// 測(cè)試邏輯
}
}到此這篇關(guān)于一篇非常好的Spring Integration 教程的文章就介紹到這了,更多相關(guān)Spring Integration 教程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- SpringIntegration消息路由之Router的條件路由與過(guò)濾功能
- Spring Integration 實(shí)現(xiàn)消息驅(qū)動(dòng)的詳細(xì)步驟
- SpringBoot integration實(shí)現(xiàn)分布式鎖的示例詳解
- 解決spring-integration-mqtt頻繁報(bào)Lost connection錯(cuò)誤問(wèn)題
- spring-integration連接MQTT全過(guò)程
- Spring?Integration概述與怎么使用詳解
- 如何使用Spring?integration在Springboot中集成Mqtt詳解
相關(guān)文章
這篇文章主要介紹了在zuulFilter中注入bean失敗的解決方案，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-07-07
idea以文本形式輸出idea目錄結(jié)構(gòu)方式
本文介紹了使用Alt+F12打開(kāi)Terminal終端,cd到項(xiàng)目路徑下,使用tree命令查看項(xiàng)目結(jié)構(gòu)的方法,并表示這是個(gè)人經(jīng)驗(yàn),僅供參考2026-05-05
SpringBoot項(xiàng)目如何連接MySQL8.0數(shù)據(jù)庫(kù)
這篇文章主要介紹了SpringBoot項(xiàng)目如何連接MySQL8.0數(shù)據(jù)庫(kù)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-11-11
Java大數(shù)據(jù)開(kāi)發(fā)Hadoop?MapReduce
MapReduce的思想核心是“分而治之”，適用于大量復(fù)雜的任務(wù)處理場(chǎng)景（大規(guī)模數(shù)據(jù)處理場(chǎng)景）Map負(fù)責(zé)“分”，即把復(fù)雜的任務(wù)分解為若干個(gè)“簡(jiǎn)單的任務(wù)”來(lái)并行處理?？梢赃M(jìn)行拆分的前提是這些小任務(wù)可以并行計(jì)算，彼此間幾乎沒(méi)有依賴關(guān)系2023-03-03
java.lang.String?類代表字符串。Java程序中所有的字符串文字（例如"abc"?）都可以被看作是實(shí)現(xiàn)此類的實(shí)例。本文主要為大家介紹了String類的常用方法，需要的可以參考一下2022-11-11
這篇文章主要介紹了Java枚舉之EnumSet詳解,使用時(shí)進(jìn)行與或運(yùn)算,但是定義多了之后,會(huì)很亂、臃腫,編寫容易出錯(cuò),EnumSet可以實(shí)現(xiàn)類似的功能,且使用起來(lái)很簡(jiǎn)潔,需要的朋友可以參考下2023-12-12
java中超過(guò)long范圍的超大整數(shù)相加算法詳解(面試高頻)
這篇文章主要介紹了java中超過(guò)long范圍的超大整數(shù)相加算法(面試高頻),本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2020-08-08
maven打包時(shí)候修改包名稱帶上git版本號(hào)和打包時(shí)間方式
這篇文章主要介紹了maven打包時(shí)候修改包名稱帶上git版本號(hào)和打包時(shí)間方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-04-04
文章總結(jié)了Groovy中的正則表達(dá)式特點(diǎn),指出其作為Java擴(kuò)展支持=~和==~操作符,轉(zhuǎn)義字符使用需特別處理,正則表達(dá)式在雙引號(hào)字符串中不需要轉(zhuǎn)義,但需雙斜線表示特殊字符,與Java相比,Groovy更簡(jiǎn)便,基于Pattern和Matcher類,用于模式匹配和捕獲子字符串2025-09-09
SpringBoot實(shí)現(xiàn)IP地址解析的示例代碼
本篇帶大家實(shí)踐在springboot項(xiàng)目中獲取請(qǐng)求的ip與詳細(xì)地址,我們的很多網(wǎng)站app中都已經(jīng)新增了ip地址顯示,具有一定的參考價(jià)值,感興趣的可以了解一下2024-01-01