事务处理 Rabbit MQ 和 Spring AMQP答案

【问题标题】：Transaction handling Rabbit MQ and Spring AMQP事务处理 Rabbit MQ 和 Spring AMQP
【发布时间】：2017-10-13 01:32:52
【问题描述】：

我在这里试图理解一些事情。我的要求是我想将记录存储在数据库中并想将消息发送到队列，然后如果它抛出一些异常我不想发送消息并且不想提交数据库事务，那么让我们用相同的方法说。现在我想到了使用 spring 事务，但是由于两个不同的资源，想到了使用 JTA 使用一些 atomikos 来同步资源 - 但我再次阅读 RMQ 不支持 2PC 或 XA 等。无论如何，我继续尝试首先尝试不添加 atomikos，我所做的只是确保我的频道交易和 @Transaction 注释得到照顾，请参见下面的示例代码 - 我没有在 pom 中添加任何特殊内容。

现在我的问题是它是如何工作的，这与 2PC 有何不同 - 使用这种方法会出现什么问题以及哪些情况会破坏最终的一致性。令人惊讶的是为什么我不必使用第三方 jta。如果这一切都很好 - 在我看来，当我们使用 rmq 和 db 使用 spring goodies 时，这似乎是最终的一致性保证！对于微服务:)

如果这不是好的解决方案，还有哪些替代方案 - 如果可能的话，我想避免工作进程等以实现最终一致性。

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setChannelTransacted(true);
    return rabbitTemplate;
}

@GetMapping
@Transactional
public void sampleEndpoint(@RequestParam boolean throwException){
    Customer a=new Customer();
    a.setCustomerName("XYZ");
    customerRepository.save(a);
    rabbitTemplate.convertAndSend("txtest","Test");
    if(throwException)
    throw new RuntimeException();
}

我在上面的例子中使用了 postgres 依赖项，使用 spring boot 1.5.7

【问题讨论】：

标签： spring-transactions spring-amqp spring-rabbit distributed-transactions

【解决方案1】：

我建议你阅读Dave Syer's article: Distributed transactions in Spring, with and without XA。

您需要在数据库事务之前启动 Rabbit 事务，以便 Rabbit 事务与 DB 事务同步，并在 DB tx 之后很快提交，如果 DB tx 回滚则回滚。

DB tx 提交成功但 Rabbit tx 回滚的可能性很小。这在文章中被称为“Best Effort 1PC”。您需要处理重复消息的可能性很小。

您没有显示所有配置，但您的 Rabbit tx 似乎会在 DB 之前提交，这可能不是您想要的。

【讨论】：

我读过文章，它的深度和信息量很大。谢谢。现在这个演示中没有太多配置，但是正如你提到的 X 事务首先启动 - 这里是如何工作的，在这种情况下，我只是简单地用 @Transactional 注释了休息控制器。我想知道如何明确启动 rabbit tx。
需要使用文中提到的ChainedTransactionManager；结合RabbitTransactionManager 和DataSourceTransactionManager。
确保两件事 - 1）在我的示例中，您提到 rabbit tx 将在 db 之前首先提交 - 原因是因为 - 在这种情况下 db tx 将首先启动（因为我们首先进行 db 操作）然后我们启动rabbit tx。现在提交以相反的顺序发生（如文章中所述），对 2）如果我使用 chainedtxmanager - 取决于我们在 xml 中定义的顺序，相应的 tx 将首先提交 - 所以如果我们将 rabbit tx mgr 首先放在 xml 中，然后是 dbtx mgr - rabbit tx 将首先启动，然后是 db - 因此 db 将首先提交，然后是 rabbit tx。
嗨，Russell，我们正在使用完全相同的模式来变异数据库并通过 RabbitMQ 发布域事件。我注意到 Tx 在消息发布部分引入了一些开销。你知道其他的替代方法吗？
不要在 cmets 中就旧答案提出新问题；它无助于社区找到问题/答案。总是提出新问题；如果相关，您可以随时参考旧问题/答案。是的，使用 RabbitMQ 的交易非常非常昂贵；如果你想同步 rabbit 和 DB 更新，就没有办法避免它。是的，它也可以用 Kafka 来完成，但它也增加了很大的开销 - 这是因为在代理确认它已收到消息（同时使用 RabbitMQ 和 Kafka）之前事务还没有完成。

【解决方案2】：

关于“它是如何工作的”问题，spring-amqp 文档中的这句话澄清了：

如果框架在发送或接收消息时已经有一个事务在进行，并且 channelTransacted 标志为真，那么消息事务的提交或回滚将被推迟到当前事务结束。如果 channelTransacted 标志为 false，则没有事务语义适用于消息传递操作（它是自动确认的）。

我的理解是，对于您的用例，您甚至不需要配置 ChainedTransactionManager 即可实现 Best Effort 1PC。 @Transactional 就足够了，兔子 tx 将在 DB tx 之后立即提交。

【讨论】：