DDD設計中的Unitwork與DomainEvent如何相容？

最近在開發過程中，遇到了一個場景，甚是棘手，在這里分享一下。希望大家腦洞大開一起來想一下解決思路。鄙人也想了一個方案拿出來和大家一起探討一下是否合理。

一、簡單介紹一下涉及的對象概念

　　工作單元：維護變化的對象列表，在整塊業務邏輯處理完全之后一次性寫入到數據庫中。

　　領域事件：領域對象本身發生某些變化時，發布的通知事件，告訴訂閱者處理相關流程。

 

二、問題來了

　　我認為最合理的領域事件的觸發點應該設計在領域對象內部，那么問題來了。當這個領域對象發生變化的上下文是一個復雜的業務場景，整個流程中會涉及到多個領域對象，所以需要通過工作單元來保證數據寫入的一致性。此時其中各個產生變化的領域對象的領域事件如果實時被發布出去，那么當工作單元在最終提交到數據庫時，如果產生了回滾，那么會導致發布了錯誤的領域事件，產生未知的后果。

 

三、問題分析

　　我能夠想到的方案是，這里領域事件的發布也通過一個類似於工作單元一樣的概念進行持續的管理，在領域對象中的發布只是做一個記錄，只有在工作單元提交成功之后，才實際發布其中所有的領域事件。

 

四、說干就干　　

實現類：

　　 public class DomainEventConsistentQueue : IDisposable
    {
        private readonly List<IDomainEvent> _domainEvents = new List<IDomainEvent>();
        private bool _publishing = false;

        public void RegisterEvent(IDomainEvent domainEvent)
        {
            if (_publishing)
            {
                throw new ApplicationException("當前事件一致性隊列已被發布，無法添加新的事件！");
            }

            if (_domainEvents.Any(ent => ent == domainEvent))  //防止相同事件被重復添加
                return;
            
            _domainEvents.Add(domainEvent);
        }

        public void Clear()
        {
            _domainEvents.Clear();
            _publishing = false;
        }

        public void PublishEvents()
        {
            if (_publishing)
            {
                return;
            }

            if (_domainEvents == null)
                return;

            try
            {
                _publishing = true;
                foreach (var domainEvent in _domainEvents)
                {
                    DomainEventBus.Instance().Publish(domainEvent);
                }
            }
            finally
            {
                Clear();
            }
        }

        public void Dispose()
        {
            Clear();
        }
    }

使用方式：

            var aggregateA = new AggregateRootA();
            var aggregateB = new AggregateRootB();

            using (var queue = new DomainEventConsistentQueue())
            {
                using (var unitwork = new SqlServerUnitOfWork(GlobalConfig.DBConnectString))
                {
 8                     aggregateA.Event(queue);
 9                     aggregateB.Event(queue);

                    var isSuccess = unitwork.Commit();
                    if (isSuccess)
                        queue.PublishEvents();
                }
            }


        public class AggregateRootA : AggregateRoot
        {
            public void Event(DomainEventConsistentQueue queue)
            {
                queue.RegisterEvent(new DomainEventA());
            }
        }

        public class AggregateRootB : AggregateRoot
        {
            public void Event(DomainEventConsistentQueue queue)
            {
                queue.RegisterEvent(new DomainEventB());
            }
        }

        public class DomainEventA : IDomainEvent
        {
            public DateTime OccurredOn()
            {
                throw new NotImplementedException();
            }

            public void Read()
            {
                throw new NotImplementedException();
            }

            public bool IsRead
            {
                get { throw new NotImplementedException(); }
            }
        }

        public class DomainEventB : IDomainEvent
        {
            public DateTime OccurredOn()
            {
                throw new NotImplementedException();
            }

            public void Read()
            {
                throw new NotImplementedException();
            }

            public bool IsRead
            {
                get { throw new NotImplementedException(); }
            }
        }

問題是解決了，但是標紅的這段代碼看着特別變扭，在產生領域事件的領域對象方法上需要增加一個與表達的業務無關的參數，這個大大破壞了DDD設計的初衷——統一語言(Ubiquitous Language)，簡潔明了的表達出每個業務行為，業務交流應與代碼保持一致。像這2行表達起來如“AggregateRootA Event DomainEventConsistentQueue”這個 DomainEventConsistentQueue其實並不是領域對象，所以其並不是領域的一部分。

 

五、陷入思考

　　這里突然想到，如果在運行中的每個線程的共享區域存儲待發布的領域事件集合，那么不就可以隨時隨地的管理當前操作上下文中的領域事件了嗎？這里需要引入ThreadLocal<T> 類。MSDN的解釋參見https://msdn.microsoft.com/zh-cn/library/dd642243(v=vs.110).aspx。該泛型類可以提供僅針對當前線程的全局存儲空間，正好能夠恰到好處的解決我們現在遇到的問題。

 

六、說改就改

實現類：

　　 public class DomainEventConsistentQueue : IDisposable
    {
        private static readonly ThreadLocal<List<IDomainEvent>> _domainEvents = new ThreadLocal<List<IDomainEvent>>();
        private static readonly ThreadLocal<bool> _publishing = new ThreadLocal<bool> { Value = false };

        private static DomainEventConsistentQueue _current;
        /// <summary>
        /// 獲取當前的領域事件一致性隊列。
        /// 由於使用了線程本地存儲變量，此處為單例模式。
        /// </summary>
        /// <returns></returns>
        public static DomainEventConsistentQueue Current()
        {
            if (_current != null)
                return _current;
            var temp = new DomainEventConsistentQueue();
            Interlocked.CompareExchange(ref _current, temp, null);
            return temp;
        }

        public void RegisterEvent(IDomainEvent domainEvent)
        {
            if (_publishing.Value)
            {
                throw new ApplicationException("當前事件一致性隊列已被發布，無法添加新的事件！");
            }

            var domainEvents = _domainEvents.Value;
            if (domainEvents == null)
            {
                domainEvents = new List<IDomainEvent>();
                _domainEvents.Value = domainEvents;
            }

            if (domainEvents.Any(ent => ent == domainEvent))  //防止相同事件被重復添加
                return;

            domainEvents.Add(domainEvent);
        }

        public void Clear()
        {
            _domainEvents.Value = null;
            _publishing.Value = false;
        }

        public void PublishEvents()
        {
            if (_publishing.Value)
            {
                return;
            }

            if (_domainEvents.Value == null)
                return;

            try
            {
                _publishing.Value = true;
                foreach (var domainEvent in _domainEvents.Value)
                {
                    DomainEventBus.Instance().Publish(domainEvent);
                }
            }
            finally
            {
                Clear();
            }
        }

        public void Dispose()
        {
            Clear();
        }
    }

使用方式：

            var aggregateA = new AggregateRootA();
            var aggregateB = new AggregateRootB();

            using (var queue = DomainEventConsistentQueue.Current())
            {
                using (var unitwork = new SqlServerUnitOfWork(GlobalConfig.DBConnectString))
                {
                    aggregateA.Event();
                    aggregateB.Event();

                    var isSuccess = unitwork.Commit();
                    if (isSuccess)
                        queue.PublishEvents();
                }
            }

        public class AggregateRootA : AggregateRoot
        {
            public void Event()
            {
                DomainEventConsistentQueue.Current().RegisterEvent(new DomainEventA());
            }
        }

        public class AggregateRootB : AggregateRoot
        {
            public void Event()
            {
                DomainEventConsistentQueue.Current().RegisterEvent(new DomainEventB());
            }
        }

        public class DomainEventA : IDomainEvent
        {
            public DateTime OccurredOn()
            {
                throw new NotImplementedException();
            }

            public void Read()
            {
                throw new NotImplementedException();
            }

            public bool IsRead
            {
                get { throw new NotImplementedException(); }
            }
        }

        public class DomainEventB : IDomainEvent
        {
            public DateTime OccurredOn()
            {
                throw new NotImplementedException();
            }

            public void Read()
            {
                throw new NotImplementedException();
            }

            public bool IsRead
            {
                get { throw new NotImplementedException(); }
            }
        }

這樣代碼看起來比之前優雅多了。這里的 DomainEventConsistentQueue.Current() 中操作的變量針對同一個線程在哪都是共享的，所以我們只管往里丟數據就好了~

 

七、方案的局限性。

　　對於執行上下文的要求較高，整個領域事件的發布必須要求在同一線程內操作。所以在使用的過程中盡量避免這種情況的發生。如果實在無法避免只能通過把DomainEventConsistentQueue 當作變量在多個線程之間傳遞了。

 

以上是個人的想法，可能有所考慮不周~ 不知道各位園子里的小伙伴們是否有處理過類似場景的經驗，歡迎留言探討，相互學習~

　　

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

 

 

▶關於作者：張帆（Zachary，個人微信號：Zachary-ZF）。堅持用心打磨每一篇高質量原創。歡迎掃描右側的二維碼~。

定期發表原創內容：架構設計丨分布式系統丨產品丨運營丨一些思考。

 

如果你是初級程序員，想提升但不知道如何下手。又或者做程序員多年，陷入了一些瓶頸想拓寬一下視野。歡迎關注我的公眾號「跨界架構師」，回復「技術」，送你一份我長期收集和整理的思維導圖。

如果你是運營，面對不斷變化的市場束手無策。又或者想了解主流的運營策略，以豐富自己的“倉庫”。歡迎關注我的公眾號「跨界架構師」，回復「運營」，送你一份我長期收集和整理的思維導圖。