DDD设计中的Unitwork与DomainEvent如何相容？

Posted 2020-07-26 Zachary_Fan

最近在开发过程中，遇到了一个场景，甚是棘手，在这里分享一下。希望大家脑洞大开一起来想一下解决思路。鄙人也想了一个方案拿出来和大家一起探讨一下是否合理。

一、简单介绍一下涉及的对象概念

　　工作单元：维护变化的对象列表，在整块业务逻辑处理完全之后一次性写入到数据库中。

　　领域事件：领域对象本身发生某些变化时，发布的通知事件，告诉订阅者处理相关流程。

 

二、问题来了

　　我认为最合理的领域事件的触发点应该设计在领域对象内部，那么问题来了。当这个领域对象发生变化的上下文是一个复杂的业务场景，整个流程中会涉及到多个领域对象，所以需要通过工作单元来保证数据写入的一致性。此时其中各个产生变化的领域对象的领域事件如果实时被发布出去，那么当工作单元在最终提交到数据库时，如果产生了回滚，那么会导致发布了错误的领域事件，产生未知的后果。

 

三、问题分析

　　我能够想到的方案是，这里领域事件的发布也通过一个类似于工作单元一样的概念进行持续的管理，在领域对象中的发布只是做一个记录，只有在工作单元提交成功之后，才实际发布其中所有的领域事件。

 

四、说干就干　　

实现类：

 1 　　 public class DomainEventConsistentQueue : IDisposable
 2     {
 3         private readonly List<IDomainEvent> _domainEvents = new List<IDomainEvent>();
 4         private bool _publishing = false;
 5 
 6         public void RegisterEvent(IDomainEvent domainEvent)
 7         {
 8             if (_publishing)
 9             {
10                 throw new ApplicationException("当前事件一致性队列已被发布，无法添加新的事件！");
11             }
12 
13             if (_domainEvents.Any(ent => ent == domainEvent))  //防止相同事件被重复添加
14                 return;
15             
16             _domainEvents.Add(domainEvent);
17         }
18 
19         public void Clear()
20         {
21             _domainEvents.Clear();
22             _publishing = false;
23         }
24 
25         public void PublishEvents()
26         {
27             if (_publishing)
28             {
29                 return;
30             }
31 
32             if (_domainEvents == null)
33                 return;
34 
35             try
36             {
37                 _publishing = true;
38                 foreach (var domainEvent in _domainEvents)
39                 {
40                     DomainEventBus.Instance().Publish(domainEvent);
41                 }
42             }
43             finally
44             {
45                 Clear();
46             }
47         }
48 
49         public void Dispose()
50         {
51             Clear();
52         }
53     }

使用方式：

 1             var aggregateA = new AggregateRootA();
 2             var aggregateB = new AggregateRootB();
 3 
 4             using (var queue = new DomainEventConsistentQueue())
 5             {
 6                 using (var unitwork = new SqlServerUnitOfWork(GlobalConfig.DBConnectString))
 7                 {
 8                     aggregateA.Event(queue);
 9                     aggregateB.Event(queue);
10 
11                     var isSuccess = unitwork.Commit();
12                     if (isSuccess)
13                         queue.PublishEvents();
14                 }
15             }
16 
17 
18         public class AggregateRootA : AggregateRoot
19         {
20             public void Event(DomainEventConsistentQueue queue)
21             {
22                 queue.RegisterEvent(new DomainEventA());
23             }
24         }
25 
26         public class AggregateRootB : AggregateRoot
27         {
28             public void Event(DomainEventConsistentQueue queue)
29             {
30                 queue.RegisterEvent(new DomainEventB());
31             }
32         }
33 
34         public class DomainEventA : IDomainEvent
35         {
36             public DateTime OccurredOn()
37             {
38                 throw new NotImplementedException();
39             }
40 
41             public void Read()
42             {
43                 throw new NotImplementedException();
44             }
45 
46             public bool IsRead
47             {
48                 get { throw new NotImplementedException(); }
49             }
50         }
51 
52         public class DomainEventB : IDomainEvent
53         {
54             public DateTime OccurredOn()
55             {
56                 throw new NotImplementedException();
57             }
58 
59             public void Read()
60             {
61                 throw new NotImplementedException();
62             }
63 
64             public bool IsRead
65             {
66                 get { throw new NotImplementedException(); }
67             }
68         }

问题是解决了，但是标红的这段代码看着特别变扭，在产生领域事件的领域对象方法上需要增加一个与表达的业务无关的参数，这个大大破坏了DDD设计的初衷——统一语言(Ubiquitous Language)，简洁明了的表达出每个业务行为，业务交流应与代码保持一致。像这2行表达起来如“AggregateRootA Event DomainEventConsistentQueue”这个 DomainEventConsistentQueue其实并不是领域对象，所以其并不是领域的一部分。

 

五、陷入思考

　　这里突然想到，如果在运行中的每个线程的共享区域存储待发布的领域事件集合，那么不就可以随时随地的管理当前操作上下文中的领域事件了吗？这里需要引入ThreadLocal<T> 类。MSDN的解释参见https://msdn.microsoft.com/zh-cn/library/dd642243(v=vs.110).aspx。该泛型类可以提供仅针对当前线程的全局存储空间，正好能够恰到好处的解决我们现在遇到的问题。

 

六、说改就改

实现类：

 1 　　 public class DomainEventConsistentQueue : IDisposable
 2     {
 3         private static readonly ThreadLocal<List<IDomainEvent>> _domainEvents = new ThreadLocal<List<IDomainEvent>>();
 4         private static readonly ThreadLocal<bool> _publishing = new ThreadLocal<bool> { Value = false };
 5 
 6         private static DomainEventConsistentQueue _current;
 7         /// <summary>
 8         /// 获取当前的领域事件一致性队列。
 9         /// 由于使用了线程本地存储变量，此处为单例模式。
10         /// </summary>
11         /// <returns></returns>
12         public static DomainEventConsistentQueue Current()
13         {
14             if (_current != null)
15                 return _current;
16             var temp = new DomainEventConsistentQueue();
17             Interlocked.CompareExchange(ref _current, temp, null);
18             return temp;
19         }
20 
21         public void RegisterEvent(IDomainEvent domainEvent)
22         {
23             if (_publishing.Value)
24             {
25                 throw new ApplicationException("当前事件一致性队列已被发布，无法添加新的事件！");
26             }
27 
28             var domainEvents = _domainEvents.Value;
29             if (domainEvents == null)
30             {
31                 domainEvents = new List<IDomainEvent>();
32                 _domainEvents.Value = domainEvents;
33             }
34 
35             if (domainEvents.Any(ent => ent == domainEvent))  //防止相同事件被重复添加
36                 return;
37 
38             domainEvents.Add(domainEvent);
39         }
40 
41         public void Clear()
42         {
43             _domainEvents.Value = null;
44             _publishing.Value = false;
45         }
46 
47         public void PublishEvents()
48         {
49             if (_publishing.Value)
50             {
51                 return;
52             }
53 
54             if (_domainEvents.Value == null)
55                 return;
56 
57             try
58             {
59                 _publishing.Value = true;
60                 foreach (var domainEvent in _domainEvents.Value)
61                 {
62                     DomainEventBus.Instance().Publish(domainEvent);
63                 }
64             }
65             finally
66             {
67                 Clear();
68             }
69         }
70 
71         public void Dispose()
72         {
73             Clear();
74         }
75     }

使用方式：

 1             var aggregateA = new AggregateRootA();
 2             var aggregateB = new AggregateRootB();
 3 
 4             using (var queue = DomainEventConsistentQueue.Current())
 5             {
 6                 using (var unitwork = new SqlServerUnitOfWork(GlobalConfig.DBConnectString))
 7                 {
 8                     aggregateA.Event();
 9                     aggregateB.Event();
10 
11                     var isSuccess = unitwork.Commit();
12                     if (isSuccess)
13                         queue.PublishEvents();
14                 }
15             }
16 
17         public class AggregateRootA : AggregateRoot
18         {
19             public void Event()
20             {
21                 DomainEventConsistentQueue.Current().RegisterEvent(new DomainEventA());
22             }
23         }
24 
25         public class AggregateRootB : AggregateRoot
26         {
27             public void Event()
28             {
29                 DomainEventConsistentQueue.Current().RegisterEvent(new DomainEventB());
30             }
31         }
32 
33         public class DomainEventA : IDomainEvent
34         {
35             public DateTime OccurredOn()
36             {
37                 throw new NotImplementedException();
38             }
39 
40             public void Read()
41             {
42                 throw new NotImplementedException();
43             }
44 
45             public bool IsRead
46             {
47                 get { throw new NotImplementedException(); }
48             }
49         }
50 
51         public class DomainEventB : IDomainEvent
52         {
53             public DateTime OccurredOn()
54             {
55                 throw new NotImplementedException();
56             }
57 
58             public void Read()
59             {
60                 throw new NotImplementedException();
61             }
62 
63             public bool IsRead
64             {
65                 get { throw new NotImplementedException(); }
66             }
67         }

这样代码看起来比之前优雅多了。这里的 DomainEventConsistentQueue.Current() 中操作的变量针对同一个线程在哪都是共享的，所以我们只管往里丢数据就好了~

 

七、方案的局限性。

　　对于执行上下文的要求较高，整个领域事件的发布必须要求在同一线程内操作。所以在使用的过程中尽量避免这种情况的发生。如果实在无法避免只能通过把DomainEventConsistentQueue 当作变量在多个线程之间传递了。

 

以上是个人的想法，可能有所考虑不周~ 不知道各位园子里的小伙伴们是否有处理过类似场景的经验，欢迎留言探讨，相互学习~

　　

作者:  Zachary
出处：https://zacharyfan.com/archives/61.html

 

 

▶关于作者：张帆（Zachary，个人微信号：Zachary-ZF）。坚持用心打磨每一篇高质量原创。欢迎扫描右侧的二维码~。

定期发表原创内容：架构设计丨分布式系统丨产品丨运营丨一些思考。

 

如果你是初级程序员，想提升但不知道如何下手。又或者做程序员多年，陷入了一些瓶颈想拓宽一下视野。欢迎关注我的公众号「跨界架构师」，回复「技术」，送你一份我长期收集和整理的思维导图。

如果你是运营，面对不断变化的市场束手无策。又或者想了解主流的运营策略，以丰富自己的“仓库”。欢迎关注我的公众号「跨界架构师」，回复「运营」，送你一份我长期收集和整理的思维导图。