TxSelect 和 TransactionScope答案

【问题标题】：TxSelect and TransactionScopeTxSelect 和 TransactionScope
【发布时间】：2012-07-31 11:39:01
【问题描述】：

最近，我一直在检查 RabbitMQ over C# 作为实现 pub/sub 的一种方式。我更习惯使用NServiceBus。 NServiceBus 通过在 TransactionScope 中登记 MSMQ 来处理事务。其他事务感知操作也可以加入同一个TransactionScope（如 MSSQL），因此一切都是真正的原子操作。下面由 NSB 引入 MSDTC 进行协调。

我看到在 RabbitMQ 的 C# 客户端 API 中有一个 IModel.TxSelect() 和 IModel.TxCommit()。这在提交之前不向交换器发送消息效果很好。这涵盖了有多个消息发送到需要原子的交换的用例。但是，有没有一种将数据库调用（比如 MSSQL）与 RabbitMQ 事务同步的好方法？

【问题讨论】：

您希望系统的吞吐量是多少？
@kzhen 我并不担心性能。不过，一致性很重要。我将使用持久的交换和队列。吞吐量不会那么高，可能每天 50-100,000 条消息。

标签： c# transactions rabbitmq

【解决方案1】：

您可以通过实现IEnlistmentNotification 接口来编写供MSDTC 使用的RabbitMQ 资源管理器。该实现在登记参与时为事务管理器提供两个阶段提交通知回调。请注意，MSDTC 价格昂贵，会大大降低您的整体性能。

RabbitMQ 资源管理器示例：

sealed class RabbitMqResourceManager : IEnlistmentNotification
{
    private readonly IModel _channel;

    public RabbitMqResourceManager(IModel channel, Transaction transaction)
    {
        _channel = channel;
        _channel.TxSelect();
        transaction.EnlistVolatile(this, EnlistmentOptions.None);
    }

    public RabbitMqResourceManager(IModel channel)
    {
        _channel = channel;
        _channel.TxSelect();
        if (Transaction.Current != null)
            Transaction.Current.EnlistVolatile(this, EnlistmentOptions.None);
    }

    public void Commit(Enlistment enlistment)
    {
        _channel.TxCommit();
        enlistment.Done();
    }

    public void InDoubt(Enlistment enlistment)
    {           
        Rollback(enlistment);
    }

    public void Prepare(PreparingEnlistment preparingEnlistment)
    {
        preparingEnlistment.Prepared();
    }

    public void Rollback(Enlistment enlistment)
    {
        _channel.TxRollback();
        enlistment.Done();
    }
}

使用资源管理器的示例

using(TransactionScope trx= new TransactionScope())
{
    var basicProperties = _channel.CreateBasicProperties();
    basicProperties.DeliveryMode = 2;

    new RabbitMqResourceManager(_channel, trx);
    _channel.BasicPublish(someExchange, someQueueName, basicProperties, someData);
    trx.Complete();
}

【讨论】：

很有趣...但是，最终我们选择了无 MSDTC 解决方案。不过，感谢您将其添加到问题中，希望其他人会发现它有用。 :)
我在我的应用程序中使用了相同的方法。但是，我发现 IEnlistmentNotification 的 Commit/Prepare 方法是在不同的线程中调用的（这似乎是 System.Transaction 中的正常行为），这会导致 RabbitMQ 出现一些问题，因为 IModel 不应跨线程使用（来自官方文档）。你试验过这个问题吗？
不，我没有。我认为他们指的是线程并发，因为通道不是线程安全的。因此，如果多个线程同时使用该通道，您将遇到问题。

【解决方案2】：

据我所知，无法协调 TxSelect/TxCommit 与 TransactionScope。

目前我采用的方法是使用带有持久消息的持久队列，以确保它们在 RabbitMQ 重新启动后仍然存在。然后，当从队列中消费时，我读取一条消息并进行一些处理，然后将一条记录插入到数据库中，一旦完成所有这些，我就会确认（nowledge）消息并将其从队列中删除。这种方法的潜在问题是消息可能最终被处理两次（例如，如果消息被提交到数据库，但在消息可以被确认之前与 RabbitMQ 的连接被断开），但是对于系统来说我们正在建设我们关注吞吐量。（我相信这被称为“至少一次”方法）。

RabbitMQ 网站确实表示使用 TxSelect 和 TxCommit 会对性能造成重大影响，因此我建议对这两种方法进行基准测试。

无论您采用哪种方式，您都需要确保您的消费者能够处理可能被处理两次的消息。

如果您还没有找到它，请查看 RabbitMQ here 的 .Net 用户指南，特别是第 3.5 节

【讨论】：

是的，我同意你写的一切。不过，在我的特殊情况下，消费者不是问题。我有一种情况，生产者可能会提交数据库状态，然后发布消息。然而，我再次完全同意消费者方面需要幂等消息。感谢您提供有关 TxSelect 和 TxCommit 性能的提示。
也许你的方法可以让生产者提交数据库行然后发送一条消息然后更新该行以说当服务器确认接收时消息已发送 (rabbitmq.com/blog/2011/02/10/introducing-publisher-confirms) 然后如果你的生产者崩溃时它重新上线，它可以查找尚未发布消息的行，然后（重新）发送它们
是的，我认为你是对的。我们将计划提交消息需要与原始数据库事务一起发送。然后让调度员将其取走并发送给 Rabbit。最后，我们将确认消息已发送。感谢您的链接！

【解决方案3】：

假设您已经为抽象 IServiceBus 实现了服务总线。我们可以假装它是引擎盖下的 rabbitmq，但它当然不需要。

当您调用 servicebus.Publish 时，您可以检查 System.Transaction.Current 以查看您是否处于事务中。如果你是并且它是一个 mssql 服务器连接的事务，你可以发布到 sql server 中的代理队列而不是发布到兔子，这将尊重你正在执行的任何数据库操作的提交/回滚（你想做一些连接这里的魔法可以避免经纪人发布将您的 txn 升级到 msdtc）

现在您需要创建一个需要读取代理队列并实际发布到rabbit的服务，这样，对于非常重要的事情，您可以保证您的数据库操作之前完成并且消息被发布到rabbit在未来的某个时候（当服务中继它时）。如果在提交代理时发生异常，这里仍然可能发生故障，但是问题的窗口会大大减少，更糟糕的情况是你最终会发布多次，你永远不会丢失一条消息。这是不太可能的，在接收之后但在提交之前下线的 sql 服务器将是一个示例，说明您何时最终会以最少的双重发布（当服务器上线时，您将再次发布）您可以智能地构建您的服务减轻一些，但除非你使用 msdtc 和它附带的所有东西 (yikes) 或构建你自己的 msdtc (yikes yikes)，否则你将遇到潜在的失败，这一切都是为了让窗口变小并且不太可能发生。

【讨论】：