一、引言

 　　在上一专题中，商家发货和用户确认收货功能引入了消息队列来实现的，引入消息队列的好处可以保证消息的顺序处理，并且具有良好的可扩展性。但是上一专题消息队列是基于内存中队列对象来实现，这样实现有一个弊端，就是一旦服务重启或出现故障时，此时消息队列中的消息会丢失，并且也记录不了日志。所以就会出现，商家发货成功后，用户并没有收到邮件通知，并且也没有日志让我们发现是否发送了邮件通知。为了解决这个问题，就需要引入一种可恢复的消息队列。目前有很多开源的消息队列都支持可恢复的，例如TibcoEms.net等。然而，微软的MSMQ也是支持这种特性的。并且MSMQ还支持分布式部署，关于MSMQ更多内容可以参考：http://www.cnblogs.com/zhili/p/MSMQ.html

 　　在本专题中将介绍为网上书店案例引入分布式消息队列和分布式缓存的实现。

二、分布式消息队列的实现

 　　MSMQ的实现原理是：消息的发送者把自己想要发送的信息放入一个容器，然后把它保存到一个系统公用空间的消息队列中，本地或异地的消息接收程序再从该队列中取出发给它的消息进行处理。所以，即使服务器突然重启，消息也会存在于系统公用空间的消息队列中，待服务器重新启动后，可以继续接受消息进行处理，从而解决上一专题存在的问题。另外，上一专题的消息队列只能被用在当前服务器中，而MSMQ支持分布式部署，不同机器都可以对MSMQ进行接收消息来处理，此时MSMQ起到一个中间件的作用。

　　在为网上书店引入分布式消息队列之前，让我们先理一下实现思路：

• 上一专题中把发货事件和收货事件发布到EventBus中，而此时需要用MsmqEventBus来替代EventBus。而MsmqEventBus的实现就很简单了，完全可以参考EventBus来实现，只是此时消息并不是进入Queue对象中，而是通过MessageQueue对象发送到系统的消息队列中。
• 而Commit方法即从系统的消息队列中出队来获得消息。再获得消息的处理器时，与上一专题的实现有点不同，因为把事件对象发送到消息队列时，需要先把事件对象先序列化为Message对象再放入消息队列中，而出队的也是消息对象，而不是上一专题中的发货事件对象。所以此时需要把出队的消息对象反序列化为对应的事件对象。

 　　有了上面的实现思路，接下来让我们一起看看MsmqEventBus的具体实现代码吧。

public class MsmqEventBus : DisposableObject, IEventBus  {       public void Publish<TMessage>(TMessage message) where TMessage : class, IEvent    {      // 将消息放入Message中Body属性进行序列化发送到消息队列中      var msmqMessage = new Message(message) { Formatter = new new[] { message.GetType() }), Label = message.GetType().ToString()};      _messageQueue.Send(msmqMessage);      _committed = false;    }    public void Publish<TMessage>(IEnumerable<TMessage> messages) where TMessage : class, IEvent    {      messages.ToList().ForEach(m =>      {        _messageQueue.Send(m);        _committed = false;      });    }    public void Commit()    {      if (this._useInternalTransaction)      {        using (var transaction = new MessageQueueTransaction())        {          try          {            transaction.Begin();            var message = _messageQueue.Receive();            if (message != null)            {              message.Formatter = new new[] { typeof(get='_blank'>string) });              var evntType = ConvertStringToType(message.Body.ToString());              var method = _publishMethod.MakeGenericMethod(evntType);              var evnt = Activator.CreateInstance(evntType);              method.Invoke(_aggregator, new object[] { evnt });              transaction.Commit();            }          }          catch          {            transaction.Abort();            throw;          }        }      }      else      {        // 从msmq消息队列中出队，此时获得的对象是消息对象        var message = _messageQueue.Receive();        if (message != null)        {          // 指定反序列化的对象，由于我们之前把对应的事件类型保存在MessageQueue中的Label属性          // 所以此时可以通过Label属性来获得目标序列化类型          message.Formatter = new new[] { ConvertStringToType(message.Label) });                    // 这样message.Body获得就是对应的事件对象，后面的处理逻辑就和EventBus一样了          var evntType =message.Body.GetType();          var method = _publishMethod.MakeGenericMethod(evntType);          method.Invoke(_aggregator, new object[] { message.Body });        }      }      _committed = true;    }  }

原标题：[.NET领域驱动设计实战系列]专题八：DDD案例：网上书店分布式消息队列和分布式缓存的实现

关键词：.NET