【说明：博主采用边写边思考的方式完成这一系列的博客，所以代码以附件为准，文中代码仅为了说明。】

结构

在学习和实现CQRS的过程中，首要参考的项目是这个【】。所以Dpfb.Cqrs中的整体结构都是参考这个例子来的，在这个基础之上添加和改进。总的来说，.Cqrs项目的整体结构如下所示：

主要包含了（命令，事件，通信，命令处理，事件处理这几个方面）。具体的角色则如下图所示：

通信中包含了事件总线和命令总线。由于查询入口（QueryEntry）如何处置暂时没想好，所以先放一个文件夹卖萌。

实现

此时的目标是实现一个最小单元的CQRS（主要是命令和事件部分）。其中命令总线和事件总线的实现比较固定，他们的职责就是为命令和事件找到对应的处理类，然后依次调用对象的接口方法，从而将一般的直接调用“打断”，同时提供各种额外操作的注入点。为了更好的控制他们如何寻找处理类，这里引入两个接口ICommandHandlerSearcher以及IEventHandlerSearcher。放在项目的Configuration名称空间下面。现在，可以为这些接口提供一个基础的实现，供测试使用。以下是接口定义：

public interface ICommand    {        Guid Id { get; set; } } public interface IEvent { Guid Id { get; set; } } public interface IEventHandler
      
        where T : IEvent { void Handle(T @event); } public interface ICommandHandler
       
         where T: ICommand { void Execute(T command); } public interface ICommandBus { void Send
        
         (T command) where T : ICommand; } public interface IEventBus { void Publish
         
          (T @event) where T : IEvent; } public interface ICommandHandlerSearcher { ICommandHandler
          
            Find
           
            () where T : ICommand; } public interface IEventHandlerSearcher { IEnumerable
            
             
              > Find
              
               () where T : IEvent; }
              
             
            
           
          
         
        
       
      

以及通信（Buses）部分的实现：

public class DpfbEventBus : IEventBus    {        void IEventBus.Publish
      
       (T @event)        {            foreach (var handler in EventHandlerSearcher.Find
       
        ())            {                handler.Handle(@event);            }        }        public IEventHandlerSearcher EventHandlerSearcher { get; set; }    }public class DpfbCommandBus : ICommandBus    {        void ICommandBus.Send
        
         (T command)        {            var handler = CommandHandlerSearcher.Find
         
          ();            handler.Execute(command);        }        public ICommandHandlerSearcher CommandHandlerSearcher { get; set; }    }
         
        
       
      

其中，两个ISearcheres的实现包含了如何寻找Handlers以及到哪里寻找Handlers，这些具体的内容，.Cqrs这层其实并不关心。处于测试的目的，将这些实现代码移到Test名称空间下。并实现本程序集内的查找：

public class TestCommandHandlerSearcher:ICommandHandlerSearcher    {        ICommandHandler
      
        ICommandHandlerSearcher.Find
       
        ()        {            var assembly = Assembly.GetCallingAssembly();            var declaredType = typeof (ICommandHandler
        
         );            var handlers = assembly.GetTypes()                .Where(i => declaredType.IsAssignableFrom(i))                .Select(i => Activator.CreateInstance(i))                .ToArray();            if (handlers.Count() != 1)                throw new Exception();            return handlers.First() as ICommandHandler
         
          ;        }    } public class TestEventHandlerSearcher : IEventHandlerSearcher    {        IEnumerable
          
           
            > IEventHandlerSearcher.Find
            
             () { var assembly = Assembly.GetCallingAssembly(); var declaredType = typeof (IEventHandler
             
              ); var handlers = assembly.GetTypes() .Where(i => declaredType.IsAssignableFrom(i)) .Select(i => Activator.CreateInstance(i)); return handlers.Cast
              
               
                >(); } }
               
              
             
            
           
          
         
        
       
      

测试

接下来添加一些实现代码，以进行测试：

public class TestEventTwo : IEvent    {        Guid IEvent.Id        {            get { throw new NotImplementedException(); }            set { throw new NotImplementedException(); }        }    }    public class TestEventOne : IEvent    {        Guid IEvent.Id        {            get { throw new NotImplementedException(); }            set { throw new NotImplementedException(); }        }    }    public class TestEventHandler : IEventHandler
      
       ,        IEventHandler
       
            {        void IEventHandler
        
         .Handle(TestEventOne @event)        {            Console.WriteLine("处理了事件eventOne");            Configuration.EventBus.Publish(new TestEventTwo());            Console.WriteLine("引发了事件eventOne");        }        void IEventHandler
         
          .Handle(TestEventTwo @event)        {            Console.WriteLine("处理了事件eventTwo");        }    }    public class TestCommandHandler : ICommandHandler
          
               {        void ICommandHandler
           
            .Execute(TestCommand command) { Console.WriteLine("获取并处理消息：" + command.Message); var eventOne = new TestEventOne(); Configuration.EventBus.Publish(eventOne); } } public class TestCommand : ICommand { Guid ICommand.Id { get { throw new NotImplementedException(); } set { throw new NotImplementedException(); } } public string Message { get; set; } } public static class Configuration { public static ICommandBus CommandBus = new DpfbCommandBus(); public static IEventBus EventBus = new DpfbEventBus(); }
           
          
         
        
       
      

其中，Buses作为单例存在于一个静态类的字段中。

最后，用一个单元测试运行：

[TestMethod]        public void TestMethod1()        {            var command = new TestCommand {Message = "国庆记得回家"};            Configuration.CommandBus.Send(command);        }

结果（倒序是正常的，总的来说，事件链的执行会是一个深度优先的调用）：

审计（以及Session）

在上个测试例子中，使用了控制台输出来表明各个事件的执行顺序。在正常开发过程中，我们总不能到处输出到控制台，就算输出了也不见得有用。所以我们使用另外的方法来实现这一点。博主第一次接触审计这个概念是在接触一个名为ABP的框架的时候【此处应有链接】，里面包含了很多的信息（用户身份，性能计数，时间等...）,当时非常震惊。于是就把ABP源码中的一个文件夹扒了下来自己用。ABP实现了Service方法的审计，用的是动态代理（Castle）。而Cqrs本身就留了无数的注入点，所以实现起来更加直观和方便。另外，博主同时参考了ABP的Session的实现，然后搞出一个土鳖版的Session。虽然比较无耻，但是对于以后ABP的理解应该会有帮助。

博主当初匆匆忙忙的撸了一个CQRS的框架，只是想试水，所以没有想很多，Auditing部分真的是直接扒下来改了改，现在开始写博客，有更多的时间思考，所以打算从头开始实现一个。首先开始抽象，Auditing是一种消息，同时还要考虑它的存放问题，所以定义了以下两个接口（由于Abudting还包含了其他方面，如web，所有接口定义移到Dpfb层）：

public interface IAuditInfo    {    } public interface IAuditStorage    {        IEnumerable
      
        Retrive();    }
      

顺便实现一个基于内存的存储：

public class MemoryAuditStorage:IAuditStorage    {        private List
      
        _inMemory = new List
       
        ();        IEnumerable
        
          IAuditStorage.Retrive()        {            return _inMemory;        }        void IAuditStorage.Save(IAuditInfo auditInfo)        {            _inMemory.Add(auditInfo);        }    }
        
       
      

接下来实现Cqrs的命令和时间的审计对象。由于事件的调用链是棵树，这里在Dpfb层引入一些必须的数据结构（见DaaStructure名称空间）。以下是针对Cqrs的AuditInfo实现：

public class CommandEventAuditInfo : ExtendedTreeNode
      
       , IAuditInfo    {        public DateTime InvokedTime { get; set; }        /// 
       
        /// 单位为毫秒        /// 
        public int InvokingDuration { get; set; }        public IDpfbSession DpfbSession { get; set; }        public Type CommandEventHandlerType { get; set; }        public Type CommandEventType { get; set; }        public Type DeclaredHandlerType { get; set; }        //todo 待扩展        public object ThreadInfo { get; set; }        public bool Stopped { get; private set; }        private Stopwatch _stopwatch = new Stopwatch();        public void Start()        {            _stopwatch.Start();        }        public void Stop()        {            _stopwatch.Stop();            InvokingDuration = (int) _stopwatch.ElapsedMilliseconds;            Stopped = true;            Current = Parent;        }        public static ConcurrentDictionary
       
         ConcurrentDic =            new ConcurrentDictionary
        
         ();        public override string ToString()        {            return PrintTree();        }        public string PrintTree()        {            var sb = new StringBuilder();            var userStr = DpfbSession != null ? DpfbSession.ToString() : "匿名用户";            var @abstract = string.Format("命令事件调用分析[{3}]：用户[{1}]引发了[{0}]，总耗时[{2}]毫秒。调用链：",                CommandEventType.Name, userStr, InvokingDuration, InvokedTime);            sb.Append(@abstract);            var recursionDepth = 1;            return PrintTree(this, ref recursionDepth, sb);        }        private string PrintTree(CommandEventAuditInfo auditInfo, ref int recursionDepth, StringBuilder sb)        {            sb.AppendLine();            var span = recursionDepth == 0                ? ""                : string.Join("", new string[recursionDepth].Select(i => "         ").ToArray());            sb.Append(span + "|---");            sb.AppendFormat("[{2}]处理[{0}]，耗时[{1}毫秒]", auditInfo.CommandEventType.Name,                auditInfo.InvokingDuration, auditInfo.CommandEventType.Name);            if (auditInfo.Children.Any())                recursionDepth++;            foreach (var commandEventAuditInfo in auditInfo.Children)            {                PrintTree(commandEventAuditInfo, ref recursionDepth, sb);            }            return sb.ToString();        }        private static CommandEventAuditInfo CreateUnstoppedOnCurrentThread()        {            Func
         
           creator = k => new CommandEventAuditInfo()            {                ThreadInfo = k            };            var threadName = Thread.CurrentThread.ManagedThreadId;            var auditInfo = ConcurrentDic.GetOrAdd(threadName, creator);            if (auditInfo.Stopped)            {                return ConcurrentDic.AddOrUpdate(threadName, creator, (k, nv) => creator((int) nv.ThreadInfo));            }            return auditInfo;        }        public static CommandEventAuditInfo StartNewForCommand
          
           (Type handlerType) where TCommand : ICommand        {            var root = CreateUnstoppedOnCurrentThread();            root.InvokedTime = DateTime.Now;            root.CommandEventHandlerType = handlerType;            root.CommandEventType = typeof (TCommand);            root.DeclaredHandlerType = typeof (ICommandHandler
           
            ); Current = root; return root; } public static CommandEventAuditInfo StartNewForEvent
            
             (Type handlerType) where TEvent : IEvent { var auditInfo = new CommandEventAuditInfo() { CommandEventType = typeof (TEvent), CommandEventHandlerType = handlerType, DeclaredHandlerType = typeof (IEventHandler
             
              ), InvokedTime = DateTime.Now }; Current.Children.Add(auditInfo); auditInfo.Parent = Current; return auditInfo; } public static CommandEventAuditInfo Root { get { return CreateUnstoppedOnCurrentThread(); } } public static CommandEventAuditInfo Current { get; private set; } }
             
            
           
          
         
        
       
      

稍作修改之后，重新运行单元测试：

[TestMethod]        public void TestAuditing()        {            var command = new TestCommand {Message = "国庆记得回家"};            Configuration.CommandBus.Send(command);            command = new TestCommand {Message = "国庆记得回家"};            Configuration.CommandBus.Send(command);            foreach (var auditInfo in Configuration.AuditStorage.Retrive())            {                Console.WriteLine(auditInfo.ToString());            }        }

这里是运行结果[事件调用链的关系有误，请移步第二篇查看]：

 至于Session，此时实现并不能表示其作用，所以只是定义了一个接口。

补丁

至此，一个可以发挥作用的Cqrs就完成了。同时也遗留下一些需要思考的问题：

【事件的继承关系】

在博主的目前实现中，是不考虑事件的继承关系的。

【事件的先后顺序】

使用Handler实现方法上的Attribute实现，这个仅仅作为辅助功能实现。逻辑上的先后顺序由事件链指定。

【CqrsAuditng的生命周期】

目前博主使用Unity作IoC，基于线程的生命周期管理，会存在并发问题（ASP.NET），需要继续考虑。

【Searchers的性能优化】

实际上调用链在编译期间就确定了，所以，可以使用ExpressionTree作缓存。

【事件/命令异步执行的支持】

在ASP.NET中，使用基于TPL的一套异步编程框架可以提高吞吐。这个可以大致这么理解：在web系统中，存在两种角色，请求入口（IIS）【标记为A】，以及请求处理者（实现代码）【标记为B】。我们先假定整个系统只有一个A。那么，在改进之前，情况是“A接到了一个请求告诉B，然后看着B把事情干完，再把结果告诉请求方”。改进之后的情况是“A接到了一个请求B，告诉B处理完事情之后给个反馈——A马上开始处理其他请求。”显然，在改进之前A的等待时间很长，改进之后A当了甩手掌柜（只等报告），等待时间很短。于是乎，A单位时间能的处理量就上去了，吞吐就上去了。然而，对于一次请求而言，时长取决于B，除非B的工作效率有所改进，不然并不会提升性能。

这些是博主的个人理解，并不求多少准确，仅希望能大致描述这么一个意思。

然而博主做了很多测试，并没有反应出这一点。详细内容，请移步这篇文章：【此处应有链接】

【Audting的配置】

这是ABP做的好的地方，也是工作量最大的地方，配置的优先级等。 

 

此篇完成时，所使用的代码：【】