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

简介: 原文:[.NET领域驱动设计实战系列]专题八:DDD案例:网上书店分布式消息队列和分布式缓存的实现一、引言    在上一专题中,商家发货和用户确认收货功能引入了消息队列来实现的,引入消息队列的好处可以保证消息的顺序处理,并且具有良好的可扩展性。
原文: [.NET领域驱动设计实战系列]专题八:DDD案例:网上书店分布式消息队列和分布式缓存的实现

一、引言

   在上一专题中,商家发货和用户确认收货功能引入了消息队列来实现的,引入消息队列的好处可以保证消息的顺序处理,并且具有良好的可扩展性。但是上一专题消息队列是基于内存中队列对象来实现,这样实现有一个弊端,就是一旦服务重启或出现故障时,此时消息队列中的消息会丢失,并且也记录不了日志。所以就会出现,商家发货成功后,用户并没有收到邮件通知,并且也没有日志让我们发现是否发送了邮件通知。为了解决这个问题,就需要引入一种可恢复的消息队列。目前有很多开源的消息队列都支持可恢复的,例如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 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.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就是真正的缓存数据,如果缓存在字典中存在的话,就直接根据键值查找出缓存数据进行返回。

  然而网上书店的本地缓存是基于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.htmlhttp://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的支持,我将会在下一专题进行介绍。

四、总结

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

  本专题所有源码下载地址:https://github.com/lizhi5753186/OnlineStore_Second/

 

目录
相关文章
|
18天前
|
开发框架 NoSQL MongoDB
C#/.NET/.NET Core开发实战教程集合
C#/.NET/.NET Core开发实战教程集合
|
2月前
|
开发框架 NoSQL .NET
利用分布式锁在ASP.NET Core中实现防抖
【9月更文挑战第5天】在 ASP.NET Core 中,可通过分布式锁实现防抖功能,仅处理连续相同请求中的首个请求,其余请求返回 204 No Content,直至锁释放。具体步骤包括:安装分布式锁库如 `StackExchange.Redis`;创建分布式锁服务接口及其实现;构建防抖中间件;并在 `Startup.cs` 中注册相关服务和中间件。这一机制有效避免了短时间内重复操作的问题。
|
2月前
|
SQL 关系型数据库 数据库
七天.NET 8操作SQLite入门到实战详细教程(选型、开发、发布、部署)
七天.NET 8操作SQLite入门到实战详细教程(选型、开发、发布、部署)
|
3月前
|
测试技术 API 开发者
.NET单元测试框架大比拼:MSTest、xUnit与NUnit的实战较量与选择指南
【8月更文挑战第28天】单元测试是软件开发中不可或缺的一环,它能够确保代码的质量和稳定性。在.NET生态系统中,MSTest、xUnit和NUnit是最为流行的单元测试框架。本文将对这三种测试框架进行全面解析,并通过示例代码展示它们的基本用法和特点。
200 8
|
3月前
|
开发框架 缓存 前端开发
实战.NET Framework 迁移到 .NET 5/6
从.NET Framework 迁移到.NET 5/6 是一次重要的技术革新,涵盖开发环境与应用架构的全面升级。本文通过具体案例详细解析迁移流程,包括评估现有应用、利用.NET Portability Analyzer 工具识别可移植代码、创建新项目、逐步迁移代码及处理依赖项更新等关键步骤。特别关注命名空间调整、JSON 序列化工具更换及数据库访问层重构等内容,旨在帮助开发者掌握最佳实践,确保迁移过程平稳高效,同时提升应用性能与可维护性。
101 2
|
3月前
|
API 开发者 Java
API 版本控制不再难!Spring 框架带你玩转多样化的版本管理策略,轻松应对升级挑战!
【8月更文挑战第31天】在开发RESTful服务时,为解决向后兼容性问题,常需进行API版本控制。本文以Spring框架为例,探讨四种版本控制策略:URL版本控制、请求头版本控制、查询参数版本控制及媒体类型版本控制,并提供示例代码。此外,还介绍了通过自定义注解与过滤器实现更灵活的版本控制方案,帮助开发者根据项目需求选择最适合的方法,确保API演化的管理和客户端使用的稳定与兼容。
132 0
|
3月前
|
缓存 NoSQL Java
惊!Spring Boot遇上Redis,竟开启了一场缓存实战的革命!
【8月更文挑战第29天】在互联网时代,数据的高速读写至关重要。Spring Boot凭借简洁高效的特点广受开发者喜爱,而Redis作为高性能内存数据库,在缓存和消息队列领域表现出色。本文通过电商平台商品推荐系统的实战案例,详细介绍如何在Spring Boot项目中整合Redis,提升系统响应速度和用户体验。
59 0
|
3月前
|
监控 Cloud Native 开发者
云端精英的.NET微服务秘籍:Azure上的创新实战演练
【8月更文挑战第28天】在现代软件开发中,微服务架构通过分解应用程序提升可维护性和扩展性。结合Azure与.NET框架,开发者能轻松打造高效且易管理的云原生微服务。首先,使用Docker容器化.NET应用,并借助Azure Kubernetes Service(AKS)或Azure Container Instances(ACI)部署。为确保高可用性和伸缩性,可利用Azure Traffic Manager负载均衡及Azure Autoscale动态调整实例数。
26 0
|
26天前
|
存储 缓存 NoSQL
数据的存储--Redis缓存存储(一)
数据的存储--Redis缓存存储(一)
60 1
|
26天前
|
存储 缓存 NoSQL
数据的存储--Redis缓存存储(二)
数据的存储--Redis缓存存储(二)
37 2
数据的存储--Redis缓存存储(二)