Spring Integration

Appearance

Spring Integration 端点权威指南 ​

用 Kotlin DSL 解锁企业集成模式，告别 XML 配置

🌟 核心概念图解 ​

一、端点类型精解 ​

1.1 通道适配器（单向通信） ​

入站适配器：数据入口（外部系统 → Spring）

kotlin
@Bean
fun fileInboundAdapter(): MessageSource<File> {
    return FileReadingMessageSource().apply {
        setDirectory(File("/input"))
        setFilter(SimplePatternFileListFilter("*.txt"))
    }
}

// [!code tip] 最佳实践：搭配 @InboundChannelAdapter 注解
@Bean
@InboundChannelAdapter(channel = "fileChannel", poller = [Poller(fixedRate = "5000")])
fun poller(): MessageSource<File> {
    /* 每5秒扫描/input目录 */
}

出站适配器：数据出口（Spring → 外部系统）

kotlin
@Bean
fun smtpOutboundAdapter(): MessageHandler {
    return Mail.outboundAdapter("smtp.gmail.com")
        .port(587)
        .credentials("user", "pass")
        .protocol("smtp")
        .to("alerts@company.com")
}

// 注意：邮件配置需开启低安全应用访问

1.2 消息网关（双向通信） ​

入站网关：外部系统主动调用（HTTP/RPC等）

kotlin
@Bean
fun httpInboundGateway(): HttpRequestHandlingMessagingGateway {
    return Http.inboundGateway("/api")
        .requestChannel("requestChannel")
        .replyChannel("replyChannel")
        .mappedRequestHeaders("Content-Type")
}

// [!code highlight] // 处理 /api 端点请求

出站网关：主动调用外部服务

kotlin
@Bean
fun restOutboundGateway(): HttpRequestExecutingMessageHandler {
    return Http.outboundGateway("https://api.payment.com/process")
        .httpMethod(HttpMethod.POST)
        .expectedResponseType(String::class.java)
}

// [!code error] // 危险：未配置超时处理！需添加.replyTimeout(5000)

二、实战模块配置 ​

2.1 文件处理端点 ​

kotlin
@Configuration
class FileIntegrationConfig {

    @Bean
    fun fileInputChannel() = DirectChannel()

    @Bean
    fun fileOutputChannel() = DirectChannel()

    // 文件监听适配器
    @Bean
    @InboundChannelAdapter(channel = "fileInputChannel", poller = [Poller(fixedDelay = "1000")])
    fun fileReader() = FileReadingMessageSource().apply {
        setDirectory(File("/inbound"))
    }

    // 文件写入网关
    @Bean
    @ServiceActivator(inputChannel = "fileOutputChannel")
    fun fileWriter() = FileWritingMessageHandler(File("/processed")).apply {
        setFileNameGenerator { 
            it.headers["filename"] as String? ?: "default.txt" 
        }
    }
}

2.2 Kafka 集成示例 ​

kotlin
@Bean
fun kafkaInbound(): KafkaMessageDrivenChannelAdapter<String, Order> {
    return Kafka.messageDrivenChannelAdapter(
        KafkaMessageListenerContainer(consumerFactory(), ContainerProperties("orders"))
    )
}

@Bean
fun kafkaOutbound(): KafkaProducerMessageHandler<String, String> {
    return Kafka.outboundChannelAdapter(producerFactory())
        .topic("processed_orders")
        .messageKeyExpression("headers['orderId']")
}

三、企业级最佳实践 ​

3.1 错误处理机制 ​

kotlin
@Bean
fun errorHandlingFlow() = IntegrationFlow.from("inputChannel")
    .handle<Any>({ _, _ -> throw RuntimeException("模拟错误") }) 
    .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
    .get()

致命陷阱

未配置死信队列（DLQ）将导致消息丢失：

kotlin
// 错误配置 ❌
.errorChannel(null)

// 正确配置 ✅
.errorChannel("dlqChannel")

3.2 性能优化方案 ​

kotlin
@Bean
fun threadPool() = ThreadPoolTaskExecutor().apply {
    corePoolSize = 5
    maxPoolSize = 10
    queueCapacity = 50
}

@Bean
fun processingFlow() = IntegrationFlow.from("orders")
    .channel(MessageChannels.executor(threadPool()))
    .handle(orderProcessor)
    .get()
kotlin
@Bean
fun backPressureChannel() = PriorityChannel(100).apply {
    setComparator(HeaderAttributeComparator("priority"))
}

四、端点选型速查表 ​

模块入站适配器出站适配器入站网关出站网关
文件系统✅✅❌✅
HTTP✅✅✅✅
Kafka✅✅✅✅
邮件✅✅❌❌
数据库✅✅❌✅

黄金法则

  1. 单向数据流 → 适配器
  2. 请求/响应模式 → 网关
  3. 高吞吐场景 → 异步通道
  4. 关键业务 → DLQ + 重试机制

五、常见问题排雷 ​

Q：消息处理超时导致阻塞？

kotlin
// 添加超时配置
@Bean
fun gateway() = IntegrationFlow.from("gatewayChannel")
    .handle(Http.outboundGateway("...")
        .replyTimeout(3000) // 新增超时设置
    )

Q：如何防止消息重复消费？

kotlin
@Bean
fun idempotentReceiver() = IdempotentReceiverInterceptor(
    MetadataStoreSelector { message -> 
        message.headers["messageId"] 
    }
)

// 注册拦截器
@Bean
fun secureFlow() = IntegrationFlow.from("input")
    .intercept(idempotentReceiver)
    ...

Q：Spring Boot 3 兼容性问题？

版本注意

gradle
// 必须使用 6.1+ 版本
implementation("org.springframework.integration:spring-integration-core:6.5.1")

“好的集成设计像中枢神经系统，端点就是传递信号的神经元” —— Spring 架构哲学

通过本指南，您已掌握：
✅ 四大端点类型的本质差异
✅ Kotlin DSL 配置最佳实践
✅ 高并发场景优化方案
✅ 生产环境故障应对策略

【实战项目模板领取】
👉 github.com/spring-integration-blueprints