协程内抛异常导致Reactor无限循环的原因排查

核心原因 #

1. 事务与协程上下文冲突：KafkaService.process方法标注了@Transactional，Spring声明式事务默认基于线程绑定上下文，但协程的挂起/恢复会切换线程，导致事务上下文被错误地重复绑定到Reactor的响应式上下文中，两者相互嵌套触发无限递归调用，最终造成栈溢出。
2. 多余的协程作用域嵌套：KafkaMapper.map方法用coroutineScope包裹了整个逻辑，但suspend fun本身已经运行在协程上下文中，awaitSingle()也不需要额外作用域，这会加剧上下文的嵌套层级，放大递归问题。
3. ACK逻辑不合理：不管处理成功与否，finally块都会调用ack.acknowledge()，导致失败消息被标记为已消费。如果后续重试机制（或死信主题发送）有问题，可能间接引发重复处理，但直接导致栈溢出的还是上下文递归。

解决步骤 #

1. 修复事务与协程的兼容性：

   • 若使用Spring Boot 3.x+，开启协程事务支持：添加@EnableCoroutineTransactionManagement注解，将@Transactional替换为@TransactionalCoroutine，或者用TransactionalOperator的协程API手动管理事务，避免线程绑定的事务上下文和协程上下文冲突。
   • 尽量在协程方法中使用协程风格的数据库操作，避免混用Reactor的Mono/Flux和await*()方法。

2. 简化协程作用域：

   • 移除KafkaMapper.map中的coroutineScope，直接在suspend fun中执行逻辑：

     suspend fun map(message: ProtoMessage): MyEntity {
         val remoteCall = clientService.someRemoteCall().awaitSingle() 
         if (remoteCall.size != 1) {
             logger.error("处理失败 :(")
             throw IllegalStateException("错误信息")
         }
         return remoteCall.first()
     }
     

3. 调整ACK逻辑：

   • 仅在消息处理完全成功时调用ack.acknowledge()；进入catch块时，先将消息发送到死信主题，确认发送成功后再ACK，或者不ACK让Kafka自动重试：

     try {
         kafkaService.process(message)
         ack.acknowledge() // 成功才确认
     } catch(e:Exception){
         // 先发送到死信主题
         kafkaService.sendToDeadLetterTopic(msg)
         ack.acknowledge() // 发送死信成功后确认，避免重复投递
     }
     

4. 检查响应式上下文配置：

   • 验证WebClient的OAuth2过滤器、观测注册器等组件是否正确处理了协程/Reactor上下文，避免上下文被无限嵌套传播。

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