在 Spring Boot 中的一个事务中组发送 kafka 消息和数据库更新

Question

我需要在一个事务中执行多个操作

• 产生卡夫卡消息
• 更新表 A
• 更新表 B

我可以发送消息并且不更新两个表（A 和 B）。我不能生成消息并更新其中一个表。

我正在尝试使用@Transactional 注释来实现我的目标

import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

 @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.SERIALIZABLE)
    public void handle(Event approvalEvent) {
        var entity = entityService.getLatestVersion(approvalEvent.getTransactionId());

        entityService.approve(entity.getTransactionId());
        logService.logApproval(entity);
        producer.send(approvalEvent);
    }

我做得对吗？

Answer 1

上述方法的问题在于，您在一个事务中与两个不同的系统（数据库和消息队列）进行交互。当在一个系统上运行成功而在另一个系统上运行失败时要处理的场景组合使解决方案变得复杂。

微服务领域有一种模式可以处理完全相同的场景。它被称为发件箱模式。

您可以阅读更多关于它的信息here。

简短的总结是您的数据库中有一个名为 outbox 的附加表，其中包含要发布到消息队列的消息。

在用于添加\更新实体的数据库事务中，您在发件箱表工具中插入一行，其中包含对该实体的操作的详细信息。

然后您从发件箱表异步读取行并通过轮询或使用更改数据捕获发布到消息队列。请参阅使用 debezium 的示例实现 here。

您的交易代码如下所示。

@Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.SERIALIZABLE)
public void handle(Event approvalEvent) {
    var entity = entityService.getLatestVersion(approvalEvent.getTransactionId());

    entityService.approve(entity.getTransactionId());
    logService.logApproval(entity);
    //Outbox is the table containing the records to be published to MQ 
    outboxRepo.save(approvalEvent);
}