使用reactive-messaging时如何正确传播JTA事务状态

问题复现代码 #

事务性REST端点实现 #

@Inject
@Channel("test")
Emitter<String> emitter;

@POST
@Transactional
public Response test() {
    emitter.send("test");
}

自定义Outgoing连接器实现 #

@ApplicationScoped
@Connector("test")
public class TestConnector implements OutgoingConnectorFactory {

    @Inject
    TransactionManager tm;

    @Override
    public SubscriberBuilder<? extends Message<?>, Void> getSubscriberBuilder(Config config) {
        return ReactiveStreams.<Message<?>>builder()
            .flatMapCompletionStage(message -> {
                tm.getTransaction(); // 当前结果为null
                return message.ack();
            })
            .ignore();
    }
}

核心根因 #

• 默认上下文传播的触发时机不匹配需求：SmallRye Reactive Messaging默认采用ON_SUBSCRIBE模式捕获上下文快照，快照生成时机为应用启动阶段、Emitter实例注入、整个反应流链路完成订阅装配的时刻。该阶段尚未进入任何HTTP请求处理流程，未绑定任何请求级别的JTA事务，因此捕获到的上下文为空。
• 此前对JtaContextProvider#currentContext的调用时机判断的偏差来源是：该方法仅会在框架指定的上下文捕获触发点被调用，之前出现的“仅在订阅阶段被执行”的现象，是默认触发模式导致的结果，而非方法本身的设计逻辑。
• 连接器侧未实现消息级别的上下文恢复：即便生产端捕获了请求上下文，默认的反应流操作符也不会在消息到达处理线程时，将消息元数据中携带的上下文快照重新绑定到当前执行线程。
• 切换为@Incoming/@Outgoing注解模式复现同样问题的原因：该模式下的反应流装配、订阅时机与Emitter模式完全一致，默认上下文传播模式相同，因此会出现相同的上下文丢失现象。

修复方案 #

1. 调整Emitter的上下文捕获触发时机
   将默认的订阅时一次性捕获，修改为单条消息发送时实时捕获当前线程的上下文。修改后的端点注入代码：

@Inject
@Channel("test")
@OnContextElect(ContextPropagation.ON_EACH_MESSAGE) // 核心配置，替换默认的ON_SUBSCRIBE模式
Emitter<String> emitter;

该配置会让框架在每次执行emitter.send()的时刻，调用JtaContextProvider#currentContext抓取当前线程绑定的JTA事务快照，并存入消息的元数据中传递到下游链路。
2. 调整连接器侧的流处理逻辑，启用上下文自动恢复
在连接器的处理流中增加上下文恢复阶段，让每条消息到达处理线程时，自动将消息元数据中存储的上下文快照重新绑定到当前线程。修改后的连接器代码：

@ApplicationScoped
@Connector("test")
public class TestConnector implements OutgoingConnectorFactory {

    @Inject
    TransactionManager tm;

    @Override
    public SubscriberBuilder<? extends Message<?>, Void> getSubscriberBuilder(Config config) {
        return ReactiveStreams.<Message<?>>builder()
            // 核心配置：自动完成消息携带的上下文快照的绑定与解绑
            .stage(SmallRyeReactiveStages.withContextCapture())
            .flatMapCompletionStage(message -> {
                // 此时tm.getTransaction()可正常获取与原REST端点绑定的JTA事务
                return message.ack();
            })
            .ignore();
    }
}

3. 补充必要依赖与事务绑定规则

• 引入smallrye-context-propagation-jta模块依赖，该模块提供JTA上下文的捕获、恢复的标准实现，缺失该依赖时框架无法识别并传递JTA事务上下文。
• 若需实现消息发送逻辑与原REST事务的原子绑定（即端点事务提交才允许发送消息、事务回滚则自动丢弃消息），需将消息的ack/nack时机绑定到原事务的完成阶段，不要在连接器处理逻辑中提前提交或回滚事务。

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