确保事件最终发布到消息队列系统的最佳方式

问题描述

请想象一下，您有如下方法:

Please, imagine you have a method like the following:

public void PlaceOrder(Order order)
{
     this.SaveOrderToDataBase(order);
     this.bus.Publish(new OrderPlaced(Order));    
}

订单存入数据库后，将一个事件发布到消息队列系统，以便同一台或另一台机器上的其他子系统处理.

After the order is saved to the database, an event is published to the message queuing system, so other subsystems on the same or another machine can process it.

但是，如果 this.bus.Publish(new OrderPlaced(Order)) 调用失败会怎样?或者机器在将订单保存到数据库后就崩溃了?该事件未发布，其他子系统无法处理它.这是无法接受的.如果发生这种情况，我需要确保最终发布该事件.

But, what happens if this.bus.Publish(new OrderPlaced(Order)) call fails? Or the machine crashes just after saving the order into the database? The event is not published and other subsystems cannot process it. This is unacceptable. If this happens I need to ensure that the event is eventually published.

我可以使用哪些可接受的策略?哪个是最好的?

What are the acceptable strategies can I use? Which is the best one?

注意:我不想使用分布式事务.

NOTE: I don't want to use distributed transactions.

Paul Sasik 非常接近，我认为我可以达到 100%.这是我的想法:

Paul Sasik is very close, and I think I can achieve 100%. This is what I thought:

首先在数据库中创建一个表Events如下:

first create a table Events in the database like the following:

CREATE TABLE Events (EventId int PRIMARY KEY)

您可能想要使用 guid 代替 int，或者您可以使用序列或标识.

You may want to use guids instead of int, or you may use sequences or identities.

然后执行以下伪代码:

open transaction
save order and event via A SINGLE transaction
in case of failure, report error and return
place order in message queue
in case of failure, report error, roll back transaction and return
commit transaction

所有事件都必须包含 EventId.当事件订阅者收到事件时，他们首先检查数据库中是否存在 EventId.

All events must include EventId. When event subscribers receive an event, they first check EventId existence in database.

这样您可以获得 100% 的可靠性，而不仅仅是 99.999%

This way you get 100% realiability, not only 99.999%

推荐答案

确保事件最终发布到消息队列系统的正确方法在这个视频 和 这篇博文

The correct way to ensure the event is eventually published to the message queuing sytem is explained in this video and on this blog post

基本上你需要将要发送到数据库的消息存储在你执行业务逻辑操作的同一个事务中，然后将消息异步发送到总线并在另一个事务中从数据库中删除消息:

Basically you need to store the message to be sent into the database in the same transaction you perform the bussines logic operation, then send the message to the bus asynchronously and delete the message from the database in another transaction:

public void PlaceOrder(Order order)
{
     BeginTransaction();
     Try 
     {
         SaveOrderToDataBase(order);
         ev = new OrderPlaced(Order);
         SaveEventToDataBase(ev);
         CommitTransaction();
     }
     Catch 
     {
          RollbackTransaction();
          return;
     }

     PublishEventAsync(ev);    
}

async Task PublishEventAsync(BussinesEvent ev) 
{
    BegintTransaction();
    try 
    {
         await DeleteEventAsync(ev);
         await bus.PublishAsync(ev);
         CommitTransaction();
    }
    catch 
    {
         RollbackTransaction();
    }

}

因为 PublishEventAsync 可能会失败，你必须稍后重试，所以你需要一个后台进程来重试失败的发送，如下所示:

Because PublishEventAsync may fail you have to retry later, so you need a background process for retrying failed sendings, something like this:

foreach (ev in eventsThatNeedsToBeSent) {
    await PublishEventAsync(ev);
}

这篇关于确保事件最终发布到消息队列系统的最佳方式的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持html5模板网！