Spring Integration

Appearance

Spring Integration Claim Check模式详解 ​

TIP

学习目标：掌握Spring Integration的Claim Check模式，理解如何通过消息指针优化大消息处理流程

概念解析：Claim Check模式 ​

问题背景 ​

在分布式系统中，大型消息体（如文件内容、图片数据等）传递会导致：

  • 性能问题：传输和处理大消息消耗网络/计算资源
  • 安全隐患：敏感数据在不必要环节暴露
  • 调试困难：日志被大消息体淹没

解决方案：Claim Check模式 ​

如同机场行李托运：

  1. 托运大件行李（存储消息）
  2. 获取行李牌（消息ID）
  3. 轻装登机（传递小ID）
  4. 凭牌取行李（按需获取完整消息）

核心组件实现 ​

1️⃣ 消息存储配置 ​

kotlin
@Configuration
class MessageStoreConfig {

    // [!code tip] 内存存储(适合测试)
    @Bean
    fun simpleMessageStore(): MessageStore {
        return SimpleMessageStore()
    }

    // [!code tip] 数据库存储(生产推荐)

    @Bean
    fun jdbcMessageStore(dataSource: DataSource): MessageStore {
        return JdbcMessageStore(dataSource)
    }
}

2️⃣ 消息入库转换器 (Claim Check In) ​

kotlin
@Bean
fun claimCheckInFlow(messageStore: MessageStore): IntegrationFlow {
    return IntegrationFlow.from("inputChannel")
        // [!code highlight] // 核心入库操作
        .claimCheckIn(messageStore)
        .channel("outputChannel")
        .get()
}

NOTE

执行过程：

  1. 接收原始消息
  2. 将消息存入MessageStore
  3. 生成唯一ID（UUID）
  4. 将ID作为新消息的负载发送

3️⃣ 消息出库转换器 (Claim Check Out) ​

kotlin
@Bean
fun claimCheckOutFlow(messageStore: MessageStore): IntegrationFlow {
    return IntegrationFlow.from("checkoutChannel")
        // [!code highlight] // 核心取件操作
        .claimCheckOut(messageStore) { it.removeMessage(true) }
        .channel("processingChannel")
        .get()
}
参数默认值说明
removeMessagefalse设为true实现单次取件
messageStore-指定消息存储源
outputChannel-结果输出通道

CAUTION

内存存储注意事项：
使用SimpleMessageStore时务必设置removeMessage=true，否则会导致内存泄漏！

关键特性详解 ​

🔒 单次取件模式 (Claim Once) ​

kotlin

@Bean
fun singleUseClaimCheckOut(messageStore: MessageStore): IntegrationFlow {
    return IntegrationFlow.from("secureChannel")
        .claimCheckOut(messageStore) { it.removeMessage(true) }
        .get()
}

类比一次性取件码：

  • 首次取出后自动销毁存储的消息
  • 防止消息被重复消费
  • 避免内存泄漏（尤其内存存储）

🧩 存储策略选择 ​

kotlin
@Bean
fun inMemoryStore(): MessageStore {
    return SimpleMessageStore() // 仅限测试环境
}
kotlin
@Bean
fun persistentStore(dataSource: DataSource): MessageStore {
    val store = JdbcMessageStore(dataSource)
    store.setTablePrefix("INT_") // 自定义表前缀
    return store
}

完整应用示例 ​

电商订单图片处理场景 ​

Kotlin实现代码 ​

kotlin
@Configuration
class OrderProcessingConfig {

    @Bean
    fun orderFlow(messageStore: MessageStore): IntegrationFlow {
        return IntegrationFlow.from("orderInput")
            // 1. 存储图片返回ID
            .enrichHeaders {
                it.header("contentType", "image/jpeg")
            }
            .claimCheckIn(messageStore)

            // 2. 处理基础订单信息
            .handle("orderService", "processMetadata")

            // 3. 按需获取图片生成缩略图
            .channel("imageProcessing")
            .claimCheckOut(messageStore)
            .handle("imageService", "generateThumbnail")

            // 4. 清理存储
            .handle("storageService", "cleanup")
            .get()
    }
}

@Service
class OrderService {
    fun processMetadata(order: Order) {
        // 处理不含图片的订单基础信息
    }
}

@Service
class ImageService {
    fun generateThumbnail(image: ByteArray) {
        // 生成缩略图逻辑
    }
}

最佳实践与常见问题 ​

✅ 推荐实践 ​

kotlin
// [!code tip] 生产环境推荐配置
@Bean
fun productionClaimCheckFlow(
    dataSource: DataSource
): IntegrationFlow {
    return IntegrationFlow.from("input")
        .claimCheckIn(jdbcMessageStore(dataSource))
        .claimCheckOut(jdbcMessageStore(dataSource)) {
            it.removeMessage(true) 
        }
        .get()
}

⚠️ 常见问题解决 ​

  1. 消息无法取出

    • 检查存储实现是否匹配（内存/DB）
    • 验证ID生成策略一致性

  2. 内存泄漏

    kotlin
    // 错误示范：未启用removeMessage
.claimCheckOut(messageStore) // 导致消息堆积

// 正确做法
.claimCheckOut(messageStore) { it.removeMessage(true) }

  3. 性能瓶颈

    • 对大于1MB的消息启用Claim Check
    • 使用分布式存储（如Redis）替代JDBC

IMPORTANT

关键决策点：

  • 测试环境：SimpleMessageStore + removeMessage=true
  • 生产环境：JdbcMessageStore/RedisMessageStore
  • 敏感数据：必须启用removeMessage=true

总结与拓展 ​

Claim Check模式的核心价值：

优势实现效果
性能优化减少网络传输量70%+
安全增强敏感数据隔离存储
系统解耦生产/消费逻辑分离

下一步学习：