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

时间 2019-11-20

标签领域驱动设计实战系列专题 ddd 案例网上书店分布式消息队列缓存实现栏目设计模式繁體版

原文原文链接

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

1、引言

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

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

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

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

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

上一专题中把发货事件和收货事件发布到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 XmlMessageFormatter(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 XmlMessageFormatter(new[] { typeof(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 XmlMessageFormatter(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;
        }
    }

　　结合上面代码的注释和前面实现思路的介绍，相信理解MsmqEventBus应该没什么问题了。接下来，咱们须要在配置文件中指定EventBus为MsmqEventBus类，另外须要在你本地专有队列中建立"OnlineStoreQueue"队列来接受消息。具体的配置文件修改成：编程

 <!--Event Bus-->
      <!--<register type="OnlineStore.Events.Bus.IEventBus, OnlineStore.Events" mapTo="OnlineStore.Events.Bus.EventBus, OnlineStore.Events">
        <lifetime type="singleton" />
      </register>-->
      
      <!--注入MsmqEventBus-->
      <register type="OnlineStore.Events.Bus.IEventBus, OnlineStore.Events"
                mapTo="OnlineStore.Events.Bus.MsmqEventBus, OnlineStore.Events">
        <lifetime type="singleton" />
        <constructor>
          <param name="path" value=".\Private$\OnlineStoreQueue" />
        </constructor>
      </register>
    </container>

　　到此，分布式消息队列的实现就完成了，具体分布式消息队列的实现效果和上一专题使用EventBus的效果是同样的，这里就再也不贴图了，你们能够自行下载源码查看。缓存

3、缓存的实现

　　在实际开发过程当中，缓存的实现是必不可少的，对于已经查询过的数据能够直接从缓存中进行读取返回给调用者，利用缓存不但能够加快响应速度，还能减轻数据库服务器的压力。在大型电子商务网站中，缓存的实现更是必不可少的功能。然而缓存的实现也有两种，一种是分布式缓存，另外一种本地缓存。在大型网站中，更多实现的是分布式缓存，对于一些少用户的企业系统，可能才会使用到本地缓存。因此在本专题中，将在网上书店案例中对这两种缓存分别进行实现。服务器

3.1 本地缓存的实现

　　首先，咱们来介绍本地缓存的实现。因为这里须要实现两种缓存，根据面向接口编程原则，咱们天然首先须要定义一个缓存接口，而后这两种具体缓存都须要实现该接口。针对缓存接口，无非是缓存数据的添加，移除，更新等操做，因此缓存接口的定义以下所示：分布式

 // 缓存接口的定义
    public interface ICacheProvider
    {
        /// <summary>
        /// 向缓存中添加一个对象
        /// </summary>
        /// <param name="key">缓存的键值</param>
        /// <param name="valueKey">缓存值的键值</param>
        /// <param name="value">缓存的对象</param>
        void Add(string key, string valueKey, object value);
        void Update(string key, string valueKey, object value);
        object Get(string key, string valueKey);
        void Remove(string key);
        bool Exists(string key);
        bool Exists(string key, string valueKey);
    }

　　在介绍本地缓存的实现以前，让咱们先来思考下本地缓存的实现思路——就是在本地缓存类中定义一个字典对象，添加缓存就是往该字典插入键值对而已，其中key就是缓存数据对应的键值，value就是真正的缓存数据，若是缓存在字典中存在的话，就直接根据键值查找出缓存数据进行返回。ide

　　然而网上书店的本地缓存是基于Enterprise Library Caching库来实现的，其实现思路和我以前介绍的思路也是同样的，只不过此时字典对象不须要咱们在类中定义，此时直接用Enterprise Library Caching库中定义的就好。有了上面的分析，本地缓存的实现理解起来也就不那么难了，具体本地缓存的实现代码以下所示：

// 表示基于Microsoft Patterns & Practices - Enterprise Library Caching Application Block的缓存机制的实现
    // 该类简单理解为对Enterprise Library Caching中的CacheManager封装
    // 该缓存实现不支持分布式缓存，更多信息参考: 
    // http://stackoverflow.com/questions/7799664/enterpriselibrary-caching-in-load-balance 
    public class EntLibCacheProvider : ICacheProvider
    {
        // 得到CacheManager实例，该实例的注册经过cachingConfiguration进行注册进去的，具体看配置文件
        private readonly ICacheManager _cacheManager = CacheFactory.GetCacheManager();

        #region ICahceProvider

        public void Add(string key, string valueKey, object value)
        {
            Dictionary<string, object> dict = null;
            if (_cacheManager.Contains(key))
            {
                dict = (Dictionary<string, object>) _cacheManager[key];
                dict[valueKey] = value;
            }
            else
            {
                dict = new Dictionary<string, object> { { valueKey, value }};
            }

            _cacheManager.Add(key, dict);
        }

        public void Update(string key, string valueKey, object value)
        {
            Add(key, valueKey, value);
        }

        public object Get(string key, string valueKey)
        {
            if (!_cacheManager.Contains(key)) return null;
            var dict = (Dictionary<string, object>)_cacheManager[key];
            if (dict != null && dict.ContainsKey(valueKey))
                return dict[valueKey];
            else
                return null;
        }

        // 从缓存中移除对象
        public void Remove(string key)
        {
            _cacheManager.Remove(key);
        }

        // 判断指定的键值的缓存是否存在
        public bool Exists(string key)
        {
            return _cacheManager.Contains(key);
        }

        // 判断指定的键值和缓存键值的缓存是否存在
        public bool Exists(string key, string valueKey)
        {
            return _cacheManager.Contains(key) &&
               ((Dictionary<string, object>)_cacheManager[key]).ContainsKey(valueKey);
        }
        #endregion 
    }

　　到此，网上书店案例中本地缓存的实现就完成了。因为本地缓存不支持分布式部署，全部的缓存都存在于单独缓存服务器中，然而，针对一些大型网站来讲，这样的实现并不适合，由于在大型网站中，须要经过多个缓存服务进行集群，须要使得缓存均匀分布在集群中的缓存服务器中。此时就须要引入分布式缓存的实现。下面让咱们具体看看分布式缓存如何在该案例中实现。

3.2 分布式缓存的实现

　　分布式缓存能够经过具体的算法把缓存均匀地分布在集群中缓存服务器中，从而用户请求的不一样数据能够路由到对应的缓存服务器中进行添加、更新或得到。分布式缓存的实现有不少种方式，能够利用Memcached和Redis开源库来实现。然而，微软的Windows Azure也提供了分布式缓存的实现，本案例中分布式缓存就是基于Windows Azure的。在对分布式缓存实现以前，须要先下载对应的dll，而后再在项目中进行引用。须要下载的dll已经包含在项目根目录下的libs文件夹下，具体须要下载的程序集截图以下所示：

　　而后在基础设施层引入这些程序集，以前就能够去实现基于Windows Azure的分布式缓存了。具体的实现代码以下所示：

// 分布式缓存，该类是对微软分布式缓存服务的封装
    // 在该案例中没用用到该缓存，可是提供在这里让你们明白微软的分布式缓存实现，并非只有memcached和Redis
    // Redis参考：http://www.cnblogs.com/ceecy/p/3279407.html 和 http://blog.csdn.net/suifeng3051/article/details/23739295
    // 关于微软分布式缓存更多介绍参考：http://www.cnblogs.com/shanyou/archive/2010/06/29/AppFabricCaching.html 
    // 和http://www.cnblogs.com/mlj322/archive/2010/04/05/1704624.html
    public class AppfabricCacheProvider : ICacheProvider
    {
        private readonly DataCacheFactory _factory = new DataCacheFactory();
        private readonly DataCache _cache;

        public AppfabricCacheProvider()
        {
            this._cache = _factory.GetDefaultCache();
        }

        #region ICacheProvider Members
        public void Add(string key, string valueKey, object value)
        {
            // DataCache中不包含Contain方法，全部用Get方法来判断对应的key值是否在缓存中存在
            var val = (Dictionary<string, object>)_cache.Get(key);
            if (val == null)
            {
                val = new Dictionary<string, object> {{ valueKey, value}};
                _cache.Add(key, val);
            }
            else
            {
                if (!val.ContainsKey(valueKey))
                    val.Add(valueKey, value);
                else
                    val[valueKey] = value;

                _cache.Put(key, val);
            }
        }

        public void Update(string key, string valueKey, object value)
        {
            Add(key, valueKey, value);
        }

        public object Get(string key, string valueKey)
        {
            return Exists(key, valueKey) ? ((Dictionary<string, object>)_cache.Get(key))[valueKey] : null;
        }

        public void Remove(string key)
        {
            _cache.Remove(key);
        }

        public bool Exists(string key)
        {
            return _cache.Get(key) != null;
        }

        public bool Exists(string key, string valueKey)
        {
            var val = _cache.Get(key);
            if (val == null)
                return false;
            return ((Dictionary<string, object>)val).ContainsKey(valueKey);
        }

        #endregion 
    }

　　经过上面的步骤，分布式缓存的实现就完成了。其实，分布式缓存和本地缓存不一样之处就在于：分布式缓存支持对应的算法能够把缓存存放在不一样的服务器上，而本地缓存只能存在于本地，而不能跨机器分布。因此对于大型网站，分布式缓存才是最好的选择，因为分布式缓存的实现和部署，无疑会增长开发和维护成本，因此对于一些小型系统（指定是单数据库服务器系统），能够考虑使用本地缓存。

　　在本案例中，因为本人没有Windows Azure环境，因此对于分布式缓存的实现也不能进行测试，因此本案例中使用的仍是本地缓存。要使缓存生效，还须要对配置文件进行修改。具体配置文件修改成：

 <unity xmlns="http://schemas.microsoft.com/practices/2010/unity">
    <sectionExtension type="Microsoft.Practices.Unity.InterceptionExtension.Configuration.InterceptionConfigurationExtension, Microsoft.Practices.Unity.Interception.Configuration" />
    <container>
      <extension type="Interception" />
      
      <!--Cache Provider--> <register type="OnlineStore.Infrastructure.Caching.ICacheProvider, OnlineStore.Infrastructure" mapTo="OnlineStore.Infrastructure.Caching.EntLibCacheProvider, OnlineStore.Infrastructure" />
<!--........-->
 </container>
</unity>

　　其实，经过上面的配置以后，缓存仍是不能生效的，由于咱们通常把缓存放在得到数据方法以前进行调用，在用户对得到数据方法调用以前，首先从缓存中进行查找，若是存在，则直接返回缓存中的数据给调用者就能够了，若是不存在再调用得到数据方法从数据库中读取，读取成功后添加到缓存中再返回给调用者。既然要在方法调用前来查找缓存，从中你是否想到了什么呢？不错，就是面向切面编程，即AOP。因此要让缓存生效，在该案例中还须要支持AOP。至于AOP的支持，我将会在下一专题进行介绍。

4、总结

　　到这里，本专题的内容就结束了，正如前面所说的，在下一专题，我将在网上书店案例中引入对AOP的支持。

　　本专题全部源码下载地址：https://github.com/lizhi5753186/OnlineStore_Second/