在讲解事件总线与 DDD 这块的时候，我有提到过 ABP vNext 有实现领域事件功能，用户可以在聚合根内部使用 AddLocalEvent(object eventData) 或 AddDistributedEvent(object eventData) 添加了领域事件。
public abstract class AggregateRoot : Entity, IAggregateRoot, IGeneratesDomainEvents, IHasExtraProperties, IHasConcurrencyStamp { // ... private readonly ICollection<object> _localEvents = new Collection<object>(); private readonly ICollection<object> _distributedEvents = new Collection<object>(); // ... // 添加本地事件。 protected virtual void AddLocalEvent(object eventData) { _localEvents.Add(eventData); } // 添加分布式事件。 protected virtual void AddDistributedEvent(object eventData) { _distributedEvents.Add(eventData); } // 获得所有本地事件。 public virtual IEnumerable<object> GetLocalEvents() { return _localEvents; } // 获得所有分布式事件。 public virtual IEnumerable<object> GetDistributedEvents() { return _distributedEvents; } // 清空聚合需要触发的所有本地事件。 public virtual void ClearLocalEvents() { _localEvents.Clear(); } // 清空聚合需要触发的所有分布式事件。 public virtual void ClearDistributedEvents() { _distributedEvents.Clear(); } }可以看到，我们在聚合内部执行任何业务行为的时候，可以通过上述的方法发送领域事件。那这些事件是在什么时候被发布的呢？
发现这几个 Get 方法有被 AbpDbContext 所调用，其实在它的内部，会在每次 SaveChangesAsync() 的时候，遍历所有实体，并获取它们的本地事件与分布式事件集合，最后由 EntityChangeEventHelper 进行触发。
public abstract class AbpDbContext<TDbContext> : DbContext, IEfCoreDbContext, ITransientDependency where TDbContext : DbContext { // ... public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default) { try { var auditLog = AuditingManager?.Current?.Log; List<EntityChangeInfo> entityChangeList = null; if (auditLog != null) { entityChangeList = EntityHistoryHelper.CreateChangeList(ChangeTracker.Entries().ToList()); } var changeReport = ApplyAbpConcepts(); var result = await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken).ConfigureAwait(false); // 触发领域事件。 await EntityChangeEventHelper.TriggerEventsAsync(changeReport).ConfigureAwait(false); if (auditLog != null) { EntityHistoryHelper.UpdateChangeList(entityChangeList); auditLog.EntityChanges.AddRange(entityChangeList); Logger.LogDebug($"Added {entityChangeList.Count} entity changes to the current audit log"); } return result; } catch (DbUpdateConcurrencyException ex) { throw new AbpDbConcurrencyException(ex.Message, ex); } finally { ChangeTracker.AutoDetectChangesEnabled = true; } } // ... protected virtual EntityChangeReport ApplyAbpConcepts() { var changeReport = new EntityChangeReport(); // 遍历所有的实体变更事件。 foreach (var entry in ChangeTracker.Entries().ToList()) { ApplyAbpConcepts(entry, changeReport); } return changeReport; } protected virtual void ApplyAbpConcepts(EntityEntry entry, EntityChangeReport changeReport) { // 根据不同的实体操作状态，执行不同的操作。 switch (entry.State) { case EntityState.Added: ApplyAbpConceptsForAddedEntity(entry, changeReport); break; case EntityState.Modified: ApplyAbpConceptsForModifiedEntity(entry, changeReport); break; case EntityState.Deleted: ApplyAbpConceptsForDeletedEntity(entry, changeReport); break; } // 添加领域事件。 AddDomainEvents(changeReport, entry.Entity); } // ... protected virtual void AddDomainEvents(EntityChangeReport changeReport, object entityAsObj) { var generatesDomainEventsEntity = entityAsObj as IGeneratesDomainEvents; if (generatesDomainEventsEntity == null) { return; } // 获取到所有的本地事件和分布式事件，将其加入到 EntityChangeReport 对象当中。 var localEvents = generatesDomainEventsEntity.GetLocalEvents()?.ToArray(); if (localEvents != null && localEvents.Any()) { changeReport.DomainEvents.AddRange(localEvents.Select(eventData => new DomainEventEntry(entityAsObj, eventData))); generatesDomainEventsEntity.ClearLocalEvents(); } var distributedEvents = generatesDomainEventsEntity.GetDistributedEvents()?.ToArray(); if (distributedEvents != null && distributedEvents.Any()) { changeReport.DistributedEvents.AddRange(distributedEvents.Select(eventData => new DomainEventEntry(entityAsObj, eventData))); generatesDomainEventsEntity.ClearDistributedEvents(); } } }