如何测试R2DBC PostgreSQL与Webflux中SELECT FOR UPDATE的并行事务

你的核心问题在于事务管理器与协程Repository不兼容，导致for update skip locked的行锁定逻辑未生效，两个并行查询拿到了相同数据。以下是具体原因和修正方案：

核心问题点 #

你使用了ReactiveTransactionManager（反应式事务管理器）来管理CoroutineCrudRepository的事务，但Spring Data协程需要**CoroutineTransactionManager**才能将事务上下文正确绑定到协程执行环境中。之前的自定义DelayedTransactionalOperator基于反应式API实现，无法适配协程上下文，导致查询根本没在事务内执行，自然不会触发行锁定。

修正后的测试方案 #

1. 依赖调整与事务管理器注入 #

确保你的测试类中注入CoroutineTransactionManager而非ReactiveTransactionManager：

@Autowired
lateinit var taskRepo: TaskRepository

@Autowired
lateinit var coroutineTxManager: CoroutineTransactionManager

2. 完整测试代码 #

import kotlinx.coroutines.async
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.springframework.transaction.annotation.Isolation
import org.springframework.transaction.annotation.Propagation
import org.springframework.transaction.reactive.TransactionDefinition
import org.springframework.transaction.reactive.transactionDefinition
import org.springframework.transaction.support.withTransaction
import java.time.Instant

@Test
fun `should select distinct tasks in parallel transactions`() = runBlocking {
    // 前置准备：插入10个NEW状态的任务，保证两个事务各能拿到5个不同数据
    val testTasks = List(10) { Task(status = "NEW", creationDate = Instant.now()) }
    taskRepo.saveAll(testTasks).collect()

    var trx1Tasks: List<Task> = emptyList()
    var trx2Tasks: List<Task> = emptyList()

    // 启动两个并行事务
    val trxJob1 = launch {
        trx1Tasks = withTransaction(coroutineTxManager, transactionDefinition {
            isolationLevel = Isolation.REPEATABLE_READ
            propagation = Propagation.REQUIRES_NEW
        }) {
            val fetchedTasks = taskRepo.selectForProcessing(5).toList()
            // 延迟提交事务，确保trx2执行查询时，trx1已锁定选中行
            delay(300)
            fetchedTasks
        }
    }

    val trxJob2 = launch {
        trx2Tasks = withTransaction(coroutineTxManager, transactionDefinition {
            isolationLevel = Isolation.REPEATABLE_READ
            propagation = Propagation.REQUIRES_NEW
        }) {
            taskRepo.selectForProcessing(5).toList()
        }
    }

    // 等待两个事务执行完成
    trxJob1.join()
    trxJob2.join()

    // 断言验证：两个列表无交集，且各自拿到5条数据
    assertEquals(5, trx1Tasks.size)
    assertEquals(5, trx2Tasks.size)
    assertTrue(trx1Tasks.intersect(trx2Tasks.toSet()).isEmpty())
}

3. 关键说明 #

• 事务上下文绑定：使用withTransaction扩展函数（Spring提供的协程事务API），确保查询在正确的事务上下文内执行，for update skip locked的行锁定逻辑生效。
• 并行执行控制：用launch启动两个独立协程事务，保证它们并行执行；trx1延迟提交，确保trx2执行查询时，trx1已经锁定了选中的行。
• 数据准备：提前插入10条NEW状态的任务，避免因数据不足导致两个事务拿到相同数据。

为什么之前的自定义Operator无效？ #

你的DelayedTransactionalOperator基于反应式TransactionalOperator实现，而协程Repository的查询是在协程上下文而非反应式流上下文执行的，两者的上下文无法互通，导致查询脱离了事务控制，for update skip locked的锁定规则根本没被触发，最终两个查询返回了相同结果。

内容的提问来源于stack exchange，提问作者dippy tippy