简单的学习，实现，领域事件，事件存储，事件溯源
为何写这篇文章
本身之前都走了弯路，觉得学习战术设计就会DDD了，其实DDD的精华在战略设计，可是对于咱们菜鸟来讲，学习一些技术概念也是挺好的
常常看到这些术语，概念太多，也想简单学习一下，记忆力比较差记录一下实现的细节
数据库
领域事件
1.领域事件是过去发生的与业务有关的事实，一但发生就不可更改，因此存储事件时只能追加
3.领域事件具备时间点的特征，全部事件链接起来会造成明显的时间轴
4.领域事件会致使目标对象状态的变化，聚合根的行为会产生领域事件，因此会改变聚合的状态
在聚合根里面维护一个领域事件的聚合，每个事件对应一个Handle，经过反射维护一个数据字典，经过事件查找到指定的Handle
领域事件实现的方式：目前看到有3种方式，MediatR，消息队列 ，发布订阅模式
eShopOnContainers 中使用的是MediatR
ENode 中使用的是EQueue，EQueue是一个纯C#写的消息队列
使用已经写好的消息队列Rabbitmq ，kafka
事件存储，事件溯源，事件快照
事件存储：存储全部聚合根里面发生过的事件
1.事件存储中能够作并发的处理，好比Command 重复，领域事件的重复
2.领域事件的重复经过聚合根Id+版本号判断，能够在数据库中创建联合惟一索引，在存储事件时检测重复，记录重复的事件，根据业务作处理
3.这里要保证存储事件与发布领域事件的一致性
如何保证存储事件与发布领域事件的一致性
先存储事件而后在发布领域事件，若是发生异常，就一直重试，一直到成功为止，也能够作必定的处理，好比重试到必定的次数，就通知，进行人工处理
我选择了CAP + Policy + Dapper
事件溯源：在事件存储中记录致使状态变化的一系列领域事件。经过持久化记录改变状态的事件，经过从新播放得到状态改变的历史。 事件回放能够返回系统到任何状态
聚合快照：聚合的生命周期各有长短，有的聚合里面有大量的事件，，事件越多加载事件以及重建聚合的执行效率就会愈来愈低，快照里面存储的是聚合
1.定时存储整个聚合根：使用定时器每隔一段时间就存储聚合到快照表中
2.定量存储整个聚合根：根据事件存储中的数量来存储聚合到快照表中
事件溯源的实现方式
1.首先咱们须要实现聚合In Memory，
2.在CommandHandler中订阅 Command命令，
建立聚合时 ，在内存中维护一个数据字典，key为：聚合根的Id，value为：聚合
修改，删除，聚合时，根据聚合根的Id，查询出聚合
若是内存中聚合不存在时：根据聚合根的Id 从聚合快照表中查询出聚合，而后根据聚合快照存储的时间，聚合根Id，查询事件存储中的全部事件，而后回放事件，获得聚合最终的状态
记录遇到的问题
因为基础很是的差，因此实现的方式都是以最简单的方式来写的，存在许多的问题，代码中有问题的地方但愿你们提出来，让我学习一下
代码的实现目前尚未写快照的部分，也没有处理EventStorage中的命令重复与聚合根+版本号重复，具体的请看汤总的ENode，里面有所有的实现
1.怎样保证存储事件，发布事件的最终一致性
2.怎么解析EventStorage中的事件，回放事件
先存储事件，当事件存储成功以后，在发布事件
存储事件失败：就一直重试，发布事件失败，使用的是CAP，CAP内部使用的是本地消息表的方式，若是发布事件失败，也一直重试，若是服务器重启了，Rabbitmq里面消息为Ack,消息没有丢，重连后会继续执行
存储事件，发布事件
/// <summary>
/// 存储聚合根中的事件到EventStorage 发布事件
/// </summary>
/// <typeparam name="TAggregationRoot"></typeparam>
/// <param name="event"></param>
/// <returns></returns>
public async Task AppendEventStoragePublishEventAsync<TAggregationRoot>(TAggregationRoot @event)
where TAggregationRoot : IAggregationRoot
{
var domainEventList = @event.UncommittedEvents.ToList();
if (domainEventList.Count == 0)
{
throw new Exception("请添加事件!");
}
await TryAppendEventStorageAsync(domainEventList).ContinueWith(async e =>
{
if (e.Result == (int)EventStorageStatus.Success)
{
await TryPublishDomainEventAsync(domainEventList).ConfigureAwait(false);
@event.ClearEvents();
}
});
}
/// <summary>
/// 发布领域事件
/// </summary>
/// <returns></returns>
public async Task PublishDomainEventAsync(List<IDomainEvent> domainEventList)
{
using (var connection =
new SqlConnection(ConnectionStr))
{
if (connection.State == ConnectionState.Closed)
{
await connection.OpenAsync().ConfigureAwait(false);
}
using (var transaction = await connection.BeginTransactionAsync().ConfigureAwait(false))
{
try
{
if (domainEventList.Count > 0)
{
foreach (var domainEvent in domainEventList)
{
await _capPublisher.PublishAsync(domainEvent.GetRoutingKey(), domainEvent).ConfigureAwait(false);
}
}
await transaction.CommitAsync().ConfigureAwait(false);
}
catch (Exception e)
{
await transaction.RollbackAsync().ConfigureAwait(false);
throw;
}
}
}
}
/// <summary>
/// 发布领域事件重试
/// </summary>
/// <param name="domainEventList"></param>
/// <returns></returns>
public async Task TryPublishDomainEventAsync(List<IDomainEvent> domainEventList)
{
var policy = Policy.Handle<SocketException>().Or<IOException>().Or<Exception>()
.RetryForeverAsync(onRetry: exception =>
{
Task.Factory.StartNew(() =>
{
//记录重试的信息
_loggerHelper.LogInfo("发布领域事件异常", exception.Message);
});
});
await policy.ExecuteAsync(async () =>
{
await PublishDomainEventAsync(domainEventList).ConfigureAwait(false);
});
}
/// <summary>
/// 存储聚合根中的事件到EventStorage中
/// </summary>
/// <returns></returns>
public async Task<int> AppendEventStorageAsync(List<IDomainEvent> domainEventList)
{
if (domainEventList.Count == 0)
{
throw new Exception("请添加事件!");
}
var status = (int)EventStorageStatus.Failure;
using (var connection = new SqlConnection(ConnectionStr))
{
try
{
if (connection.State == ConnectionState.Closed)
{
await connection.OpenAsync().ConfigureAwait(false);
}
using (var transaction = await connection.BeginTransactionAsync().ConfigureAwait(false))
{
try
{
if (domainEventList.Count > 0)
{
foreach (var domainEvent in domainEventList)
{
EventStorage eventStorage = new EventStorage
{
Id = Guid.NewGuid(),
AggregateRootId = domainEvent.AggregateRootId,
AggregateRootType = domainEvent.AggregateRootType,
CreateDateTime = domainEvent.CreateDateTime,
Version = domainEvent.Version,
EventData = Events(domainEvent)
};
var eventStorageSql =
$"INSERT INTO EventStorageInfo(Id,AggregateRootId,AggregateRootType,CreateDateTime,Version,EventData) VALUES (@Id,@AggregateRootId,@AggregateRootType,@CreateDateTime,@Version,@EventData)";
await connection.ExecuteAsync(eventStorageSql, eventStorage, transaction).ConfigureAwait(false);
}
}
await transaction.CommitAsync().ConfigureAwait(false);
status = (int)EventStorageStatus.Success;
}
catch (Exception e)
{
await transaction.RollbackAsync().ConfigureAwait(false);
throw;
}
}
}
catch (Exception e)
{
connection.Close();
throw;
}
}
return status;
}
/// <summary>
/// AppendEventStorageAsync异常重试
/// </summary>
public async Task<int> TryAppendEventStorageAsync(List<IDomainEvent> domainEventList)
{
var policy = Policy.Handle<SocketException>().Or<IOException>().Or<Exception>()
.RetryForeverAsync(onRetry: exception =>
{
Task.Factory.StartNew(() =>
{
//记录重试的信息
_loggerHelper.LogInfo("存储事件异常", exception.Message);
});
});
var result = await policy.ExecuteAsync(async () =>
{
var resulted = await AppendEventStorageAsync(domainEventList).ConfigureAwait(false);
return resulted;
});
return result;
}
/// <summary>
/// 根据DomainEvent序列化事件Json
/// </summary>
/// <param name="domainEvent"></param>
/// <returns></returns>
public string Events(IDomainEvent domainEvent)
{
ConcurrentDictionary<string, string> dictionary = new ConcurrentDictionary<string, string>();
//获取领域事件的类型（方便解析Json）
var domainEventTypeName = domainEvent.GetType().Name;
var domainEventStr = JsonConvert.SerializeObject(domainEvent);
dictionary.GetOrAdd(domainEventTypeName, domainEventStr);
var eventData = JsonConvert.SerializeObject(dictionary);
return eventData;
}
解析EventStorage中存储的事件
服务器
public async Task<List<IDomainEvent>> GetAggregateRootEventStorageById(Guid AggregateRootId)
{
try
{
using (var connection = new SqlConnection(ConnectionStr))
{
var eventStorageList = await connection.QueryAsync<EventStorage>($"SELECT * FROM dbo.EventStorageInfo WHERE AggregateRootId='{AggregateRootId}'");
List<IDomainEvent> domainEventList = new List<IDomainEvent>();
foreach (var item in eventStorageList)
{
var dictionaryDomainEvent = JsonConvert.DeserializeObject<Dictionary<string, string>>(item.EventData);
foreach (var entry in dictionaryDomainEvent)
{
var domainEventType = TypeNameProvider.GetType(entry.Key);
if (domainEventType != null)
{
var domainEvent = JsonConvert.DeserializeObject(entry.Value, domainEventType) as IDomainEvent;
domainEventList.Add(domainEvent);
}
}
}
return domainEventList;
}
}
catch (Exception ex)
{
throw;
}
注意事项
1.事件没持久化就表明事件还没发生成功，事件存储可能失败，必须先存储事件，在发布事件，保证存储事件与发布事件一致性
1.使用事件驱动，必需要作好冥等的处理
2.若是业务场景中有状态时：经过状态来控制
3.新建一张表，用来记录消费的信息，消费端的代码里面，根据惟一的标识，判断是否处理过该事件
4.Q端的任何更新都应该把聚合根ID和事件版本号做为条件,Q端的更新不用遵循聚合的原则，可使用最简单的方式处理
5.仓储是用来重建聚合的，它的行为和集合同样只有Get ，Add ，Delete
6.DDD不是技术，是思想，核心在战略模块，战术设计是实现的一种选择，战略设计，须要面向对象的分析能力，职责分配，深层次的分析业务
并发
感谢
虽然学习DDD的时间不短了，感受仍是在入门阶段，在学习的过程当中有许多的不解，常常问ENode群里面的大佬，也常常@汤总，谢谢你们的帮助与解惑。相关文章
- 1. DDD~领域事件与事件总线
- 2. 领域事件、事件风暴
- 3. 事件溯源Event Sourcing
- 4. Event Sourcing 事件溯源
- 5. 第22章 事件溯源
- 6. 使用领域事件
- 7. DDD(六)--领域事件
- 8. 识别领域事件
- 9. 领域服务、领域事件
- 10. jQuery事件-表单事件
- 更多相关文章...
- • C# 事件（Event） - C#教程
- • jQuery Mobile 事件 - jQuery Mobile 教程
- • 常用的分布式事务解决方案
- • Docker容器实战(七) - 容器眼光下的文件系统
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. DDD~领域事件与事件总线
- 2. 领域事件、事件风暴
- 3. 事件溯源Event Sourcing
- 4. Event Sourcing 事件溯源
- 5. 第22章 事件溯源
- 6. 使用领域事件
- 7. DDD(六)--领域事件
- 8. 识别领域事件
- 9. 领域服务、领域事件
- 10. jQuery事件-表单事件