Spring Integration

Appearance

Spring Integration 事务支持详解 ​

引言 ​

在分布式系统中，事务管理是保证数据一致性的关键机制。Spring Integration 提供了强大的事务支持，帮助开发者处理消息流中的复杂事务场景。本教程将通过通俗易懂的方式讲解事务在消息流中的应用，并提供实用的 Kotlin 实现示例。

一、理解消息流中的事务机制 ​

1.1 消息流的启动方式 ​

Spring Integration 中的消息流可通过六种机制启动：

1.2 事务处理差异 ​

启动类型特点事务配置方式
用户进程启动由外部调用触发使用 Spring 标准事务支持
守护进程启动自动定时触发需要显式配置事务支持

最佳实践

用户进程启动的流可直接使用 @Transactional 注解：

kotlin
@Service
class OrderService {

    @Transactional
    fun processOrder(order: Order) {
        // 业务处理逻辑
    }
}

二、轮询器事务支持 ​

2.1 轮询器事务配置 ​

轮询器(Poller)是守护进程启动的典型代表，需要显式配置事务：

kotlin
@Configuration
class PollerConfig {

    @Bean
    fun transactionPoller(): PollerMetadata {
        return Pollers.fixedRate(1000)
            .maxMessagesPerPoll(1)
            .transactional(transactionManager())
            .get()
    }

    @Bean
    fun transactionManager(): PlatformTransactionManager {
        return DataSourceTransactionManager(dataSource())
    }

    // 高亮事务管理器配置
    @Bean
    fun dataSource(): DataSource {
        // 数据源配置
    }
}

2.2 使用事务建议链 ​

当需要多个建议时，可使用建议链：

kotlin
@Bean
fun adviceChainPoller(): PollerMetadata {
    return Pollers.fixedRate(5000)
        .advice(transactionAdvice(), loggingAdvice())
        .get()
}

@Bean
fun transactionAdvice(): Advice {
    return TransactionInterceptor(transactionManager(), DefaultTransactionAttribute())
}

IMPORTANT

事务建议必须在建议链中第一个配置，确保事务最先开始最后结束

三、事务边界管理 ​

3.1 线程与事务边界 ​

事务上下文绑定到当前线程，线程切换会破坏事务边界：

3.2 保持事务边界方案 ​

使用事务型通道保持事务连续性：

kotlin
@Bean
fun transactionalChannel(): PollableChannel {
    return QueueChannel(
        MessageStoreFactory().createMessageStore("transactionalStore")
    ).apply {
        setTransactionManager(transactionManager())
    }
}

关键点

  • 使用 ExecutorChannel 会破坏事务边界
  • QueueChannel 配合事务型 MessageStore 可维持事务
  • JMS 等外部系统自带事务支持

四、事务同步机制 ​

4.1 事务同步应用场景 ​

在文件处理流程中，根据事务结果移动文件：

kotlin
@Bean
fun fileInboundAdapter(): MessageSource<File> {
    val adapter = FileReadingMessageSource().apply {
        setDirectory(File("/input"))
    }

    return IntegrationFlows.from(adapter) {
        it.poller(Pollers.fixedDelay(1000)
            .transactional(transactionManager())
            .transactionSynchronizationFactory(syncFactory())
        )
    }.channel("processChannel")
    .get()
}

@Bean
fun syncFactory(): TransactionSynchronizationFactory {
    return DefaultTransactionSynchronizationFactory(
        ExpressionEvaluatingTransactionSynchronizationProcessor().apply {
            setAfterCommitExpression("payload.renameTo(new File('/success/' + payload.name))")
            setAfterRollbackExpression("payload.renameTo(new File('/failed/' + payload.name))")
        }
    )
}

4.2 事务同步处理流程 ​

五、伪事务处理 ​

5.1 伪事务应用场景 ​

当流程中没有真实事务资源但仍需事务语义时：

kotlin
@Bean
fun pseudoTransactionManager(): PlatformTransactionManager {
    return PseudoTransactionManager()
}

@Bean
fun ftpOutboundFlow(): IntegrationFlow {
    return IntegrationFlow.from("ftpChannel")
        .handle(Ftp.outboundAdapter(sessionFactory(), FileExistsMode.REPLACE)
            .apply { remoteDirectory = "/upload" })
        .get()
}

5.2 伪事务配置 ​

kotlin
@Bean
fun filePoller(): PollerMetadata {
    return Pollers.fixedRate(5000)
        .transactional(pseudoTransactionManager())
        .transactionSynchronizationFactory(syncFactory())
        .get()
}

注意

伪事务不会真正提交或回滚资源，仅提供事务同步钩子：

  • 成功流程：调用 beforeCommit/afterCommit
  • 失败流程：调用 afterRollback

六、响应式事务支持 ​

6.1 响应式事务配置 ​

Spring Integration 5.3+ 支持响应式事务：

kotlin
@Bean
fun reactiveTransactionManager(): ReactiveTransactionManager {
    return ReactiveMongoTransactionManager(reactiveMongoDatabaseFactory)
}

@Bean
fun reactiveFlow(): IntegrationFlow {
    return IntegrationFlow.from(
        MongoDb.reactiveInboundChannelAdapter(reactiveMongoTemplate)
            .collectionName("data")
            .entityClass(Data::class.java)
            .expectSingleResult(true)
    )
    .handle({ payload, _ ->
        Mono.fromCallable {
            // 业务处理
        }.subscribeOn(Schedulers.boundedElastic())
    }, { it.transactional(reactiveTransactionManager()) })
    .get()
}

CAUTION

响应式事务要求：

  • 使用 ReactiveTransactionManager
  • 消息处理返回 Mono 或 Flux
  • 正确配置订阅调度器

七、常见问题解决方案 ​

7.1 事务不生效问题排查 ​

问题现象可能原因解决方案
事务未回滚异常被捕获未传播确保异常抛出到事务边界
部分操作未回滚使用了非事务资源使用事务型资源或伪事务
事务边界断开线程切换未同步使用事务型通道连接

7.2 事务同步最佳实践 ​

kotlin
// 高亮关键配置
@Bean
fun safeSyncFactory(): TransactionSynchronizationFactory {
    return DefaultTransactionSynchronizationFactory(
        ExpressionEvaluatingTransactionSynchronizationProcessor().apply {
            setBeforeCommitExpression("#root") // 访问原始消息
            setAfterCommitExpression("payload.delete()") // 事务成功后删除文件
            setAfterRollbackExpression("T(java.nio.file.Files).move(payload.toPath(), Paths.get('/error'))")
        }
    )
}
完整的事务同步配置示例
kotlin
@Configuration
class TransactionSyncConfig {

    @Bean
    fun fileProcessingFlow(): IntegrationFlow {
        return IntegrationFlow.from(
            Files.inboundAdapter(Paths.get("/input"))
                .filter(FileListFilter { true }),
            { e -> e.poller(Pollers.fixedDelay(1000)
                .transactional(transactionManager())
                .transactionSynchronizationFactory(syncFactory()))
            })
            .handle(File::class) { file, _ ->
                // 文件处理逻辑
            }
            .get()
    }

    @Bean
    fun syncFactory(): TransactionSynchronizationFactory {
        return DefaultTransactionSynchronizationFactory(
            ExpressionEvaluatingTransactionSynchronizationProcessor().apply {
                setBeforeCommitExpression(
                    "T(java.nio.file.Files).size(payload.toPath())")
                setAfterCommitExpression(
                    "payload.renameTo(new java.io.File('/processed/' + payload.name))")
                setAfterRollbackExpression(
                    "payload.renameTo(new java.io.File('/failed/' + payload.name))")
            }
        )
    }
}

总结 ​

Spring Integration 提供了灵活的事务支持方案：

  1. 轮询器事务：守护进程启动流的基础事务支持
  2. 事务同步：在事务生命周期关键点执行操作
  3. 伪事务：无真实事务资源时的替代方案
  4. 响应式事务：响应式编程模型的事务支持

通过合理使用这些特性，可以构建出健壮可靠的集成解决方案，确保数据在各种场景下的一致性。