分布式数据库事务中的RabbitMQ和交付保证

Question

我试图了解在分布式数据库事务的上下文中处理 RabbitMQ 交付的正确模式是什么。

为了简单起见，我将用伪代码来说明我的想法，但我实际上是使用 Spring AMQP 来实现这些想法的。

任何像

void foo(message) {
   processMessageInDatabaseTransaction(message);
   sendMessageToRabbitMQ(message);
}

当我们到达时 sendMessageToRabbitMQ() processMessageInDatabaseTransaction()已成功将其更改提交到数据库，或者在到达消息发送代码之前已抛出异常。

我知道对于 sendMessageToRabbitMQ()我可以用 Rabbit transactions或 publisher confirms保证兔子收到我的消息。

我的兴趣是了解当事情发生时会发生什么，即当数据库事务成功时，但在一定时间后确认没有到达(使用发布者确认)或 Rabbit 事务未能提交(使用 Rabbit 事务)。

一旦发生这种情况，保证我的消息传递的正确模式是什么？

当然，在开发了幂等消费者之后，我考虑过可以重试发送消息，直到 Rabbit 确认成功:

void foo(message) {
   processMessageInDatabaseTransaction(message);
   retryUntilSuccessFull {
      sendMessagesToRabbitMQ(message);
   }
}

但是这种模式有几个我不喜欢的缺点，首先，如果故障持续时间较长，我的线程将开始在这里阻塞，我的系统最终将变得无响应。其次，如果我的系统崩溃或关闭会发生什么？我永远不会传递这些消息，因为它们会丢失。

所以，我想，好吧，我必须首先将我的消息写入数据库，处于待处理状态，然后从那里发布我的待处理消息:

void foo(message) {
   //transaction commits leaving message in pending status
   processMessageInDatabaseTransaction(message);
}

@Poller(every="10 seconds")
void bar() {
   for(message in readPendingMessagesFromDbStore()) {
      sendPendingMessageToRabbitMQ(message);
      if(confirmed) {
          acknowledgeMessageInDatabase(message); 
      }
   }
}

如果我无法确认数据库中的消息，可能会多次发送消息。

但是现在我引入了其他问题:

需要从数据库执行 I/O 以发布一条消息，即 99% 的时间将立即成功发布，而无需检查数据库。

使轮询器更接近实时传递的困难，因为现在我已经增加了消息发布的延迟。

也许还有其他复杂情况，例如保证按顺序传递事件、轮询器执行相互交错、多个轮询器等。

然后我想好了，我可以让这更复杂一点，我可以从数据库发布，直到 catch 事件的实时流，然后实时发布，即维护大小为 b(循环缓冲区)的缓冲区为我根据页面阅读检查该消息是否在缓冲区中。如果是这样，则切换到实时订阅。

到目前为止，我意识到如何正确地做到这一点并不完全清楚，因此我得出结论，我需要学习解决这个问题的正确模式是什么。

那么，是否有人对正确执行此操作的正确方法有任何建议？

Answer 1

当 Rabbit 未能收到消息时(无论出于何种原因，但根据我的经验，仅是因为服务关闭或不可用)您应该能够捕获错误。此时，您可以记录该尝试以及任何后续失败的尝试，以便在 Rabbit 再次可用时重试。执行此操作的最快方法是将消息详细信息记录到文件中，然后迭代以重新发送 适当时 .

只要您拥有该文件，就不会丢失消息。

一旦消息在 Rabbit 中，并且您对架构的其余部分有信心，就应该可以安全地假设消息将在它们应该出现的地方结束，并且在您的最后不需要做进一步的持久性工作。