.NET分布式事务处理总结【下】 - 包含MSMQ的分布式事务处理

时间 2019-11-24

原文原文链接

转自：http://www.cnblogs.com/daxnet/archive/2011/03/15/1984995.htmlhtml

.NET直接提供对MSMQ的访问支持，只须要添加System.Messaging程序集引用便可方便地操做MSMQ。MSMQ支持两种事务处理模式：内部事务处理以及基于MS-DTC的分布式事务处理。数据库

MSMQ的内部事务处理框架

MSMQ的内部事务处理是指，仅采用MSMQ自己提供的事务处理机制完成事务处理。好比，假设有一系列的消息须要发布到MSMQ，那么，就能够启动一个内部事务，确保这些消息的发布过程是一个原子操做。要使用MSMQ的内部事务处理机制，在建立消息队列的时候，就须要勾选“事务性”选项，以下图所示：分布式

首先，须要建立一个MessageQueueTransaction的对象，并使用Begin调用以启动MSMQ的内部事务处理。而后，在MessageQueue的Send方法中，使用Send(object, MessageQueueTransaction)的重载函数发送消息，将建立好的MessageQueueTransaction对象做为第二个参数传递给Send方法；在完成全部消息的发送以后，使用MessageQueueTransaction对象的Commit方法提交事务。若是在发送消息的过程当中遇到问题，则使用MessageQueueTransaction对象的Abort调用回滚事务。请参见下面的示例代码：函数

using (MessageQueue messageQueue = new MessageQueue(@".\private$\TPCDemoQueue",

    false, false, QueueAccessMode.SendAndReceive))

```
{
```

    MessageQueueTransaction trans = new MessageQueueTransaction();

```
    try
```
```
    {
```
```
        trans.Begin();
```
```
        for (int i = 0; i < 5; i++)
```
```
        {
```

            messageQueue.Send(new Message(i), trans);

```
        }
```
```
        trans.Commit();
```
```
    }
```
```
    catch
```
```
    {
```
```
        trans.Abort();
```
```
    }
```
```
    messageQueue.Close();
```
```
}
```

注意：若是你的消息队列在建立的时候没有设置“事务性”选项，那么，在完成消息队列的建立之后，你将没法修改该选项。更糟糕的是，在非事务性队列上执行上面的代码，则没法将消息发布到消息队列上，框架自己也不会提示任何错误信息，指示消息并未发布成功。工具

在分布式事务处理中使用MSMQ性能

在分布式事务处理的上下文中（好比，.NET 2.0+的TransactionScope中），上面所提到的MessageQueueTransaction将毫无用处，也就是说，MessageQueueTransaction与分布式事务处理毫无关系。你所要作的是，用正常的方式初始化一个MessageQueue的实例，而后，调用Send方法发布消息，在发布消息的时候，经过设置MessageQueueTransactionType的值来告诉MessageQueue，目前正处于一个分布式事务的上下文中。因而，你须要使用Send/Receive的重载方法：Send(object, MessageQueueTransactionType)以及Receive(MessageQueueTransactionType)。以下：spa

using (TransactionScope transaction = new TransactionScope())

```
{
```

    Message inputMsg = inputQueue.Receive(MessageQueueTransactionType.Automatic);

```
    // do some work
```
```
    transaction.Complete();
```
```
}
```

注意：对于一些生命周期相对较长的事务处理，好比，假设你的用例是这样的：你首先须要从一个消息队列中得到消息，而后更新你的数据库记录，那么你的代码可能会是这样的：3d

using (TransactionScope transaction = new TransactionScope())

```
{
```

    using (MessageQueue someQueue = new MessageQueue("<queue connection>"))

```
    {
```

        Message msg = someQueue.Receive();

```
        // do something else
```
```
    }
```
```
    transaction.Complete();
```
```
}
```

这样作实际上是不对的！由于Receive方法是一种同步调用，若是消息队列中根本没有任何内容，那么Receive调用就会被阻塞，直到消息队列中出现新的消息。这就意味着你的分布式事务一直都是处于开启的状态，并且可能因为等待时间过长而致使超时，最终致使一个MessageQueueException。htm

正确的作法是，在MessageQueue上使用BeginPeek调用（注意：不是BeginReceive方法，由于BeginReceive方法并非用来处理事务性队列的），而后订阅PeekComplete事件，在事件处理过程当中，再使用TransactionScope以及Receive等方法实现消息的获取。例如：

MessageQueue inputQueue = new MessageQueue("<queue connection>");

```
inputQueue.PeekCompleted += (s, e) =>
```
```
    {
```

        using (TransactionScope transaction = new TransactionScope())

```
        {
```

            Message inputMsg = inputQueue.Receive(MessageQueueTransactionType.Automatic);

```
            // do some work
```
```
            transaction.Complete();
```
```
        }
```
```
        inputQueue.BeginPeek();
```
```
    };
```
```
inputQueue.BeginPeek();
```

最后再提醒一下，就是若是你所要作的事情仅限于与MSMQ打交道，那么只要使用MSMQ的内部事务处理机制就能够了，毕竟使用分布式事务处理会涉及到MS-DTC，从而形成过大的系统开销，影响性能。