微服务实战(六)：落地微服务架构到直销系统(事件存储)

用户1910585

发布于 2018-09-29 15:36:36

5760

发布于 2018-09-29 15:36:36

在CQRS架构中，一个比较重要的内容就是当命令处理器从命令队列中接收到相关的命令数据后，通过调用领域对象逻辑，然后将当前事件的对象数据持久化到事件存储中。主要的用途是能够快速持久化对象此次的状态，另外也可以通过未来最终一致性的需求，通过事件数据将对象还原到一个特定的状态，这个状态通常是通过对象事件的版本来进行还原的。

要实现一个事件存储的框架，我们通常需要实现以下几个方面：

1.对象事件的存储表

我们通常将对象某个变化的事件数据存储到数据库的表中，通常采用关系型数据库进行存储，这里使用SQL Server。

CREATE TABLE [dbo].[DomainCommandAndEventObject](
	[Id] [uniqueidentifier] NULL,
	[AggregationRootId] [uniqueidentifier] NULL,
	[AssemblyQualifiedAggreateRooType] [nvarchar](500) NULL,
	[AssemblyQualifiedCommandAndEventType] [nvarchar](500) NULL,
	[CreateDate] [datetime] NULL,
	[Version] [int] NULL,
	[Data] [varbinary](max) NULL
)

AggregationRootId是当前聚合根对象的Id；AssemblyQualifiedAggreateRooType是当前聚合根对象的FQDN名，在C#代码中对应名称空间+类名(例如：Order.Domain.POCOModels.Orders, Order.Domain, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null)；AssemblyQualifiedCommandAndEventType是操作当前聚合根的事件类型的FQDN名字，在C#代码中对应名称空间+类名(例如：Events.OrderCommands.CreateOrderCommand, Events, Version=1.0.0.0, Culture=neutral, PublicKeyToken=null)，Version对应的是针对某个聚合根当前事件操作的版本，通常对一个聚合根进行操作，版本就加1，Data则包括当前事件操作后，对象的当前状态数据。

2.重构Event用以支持存储

前面文章实现的事件只是用于标识消息，在事件需要存储时，我们需要更多的属性，包括聚合根ID，聚合根类型，操作聚合根的事件类型，版本号等。

   public interface IEvent
    {
        Guid Id { get; set; }
        DateTime CreateDate { get; set; }
        Guid AggregationRootId { get; set; }
        string AssemblyQualifiedAggreateRooType { get; set; }
        string AssemblyQualifiedCommandAndEventType { get; set; }
        int Version { get; set; }
    }

public class BaseEvent : IEvent
    {
        public Guid Id { get; set; }
        public DateTime CreateDate { get; set; }
        public Guid AggregationRootId { get; set; }
        public string AssemblyQualifiedAggreateRooType { get; set; }
        public string AssemblyQualifiedCommandAndEventType { get; set; }
        public int Version { get; set; }

        public BaseEvent()
        {
            this.Id = Guid.NewGuid();
            this.CreateDate = DateTime.Now;
        }
    }

3.实现存储的事件对象

其实这里要实现的就是将事件和事件对象之间做相互的转换，用于未来存储事件或将事件反序列化成事件对象进行使用。

 public class EventObject:BaseEvent
    {        
        public byte[] Data { get; set; }
        public static EventObject FromDomainEvent(IEvent idomainevent)
        {
            var domaineventobject = new EventObject();
            domaineventobject.Id = idomainevent.Id;
            domaineventobject.CreateDate = idomainevent.CreateDate;
            domaineventobject.Version = idomainevent.Version;
            domaineventobject.AggregationRootId = idomainevent.AggregationRootId;
            domaineventobject.AssemblyQualifiedAggreateRooType = idomainevent.AssemblyQualifiedAggreateRooType;
            domaineventobject.AssemblyQualifiedCommandAndEventType = idomainevent.AssemblyQualifiedCommandAndEventType;
            domaineventobject.Data = XmlObjectSerializer.Serialize(idomainevent);
            return domaineventobject;
        }
        public  IEvent ToDomainEvent()
        {            
            Type type = Type.GetType(this.AssemblyQualifiedAggreateRooType);
            var domainevent = (IEvent)XmlObjectSerializer.Deserialize(type, this.Data);
            domainevent.Id = this.Id;
            return domainevent;
        }
    }

FromDomainEvent方法就是将事件信息转换为以后要存储的事件对象，ToDomainEvent就是将事件对象转换为事件。

4.实现事件存储

实现事件存储就是将领域事件对象存储到我们前面创建的数据库表中。为了能够快速存储，我们并不采用ORM框架，而是直接使用ADO.NET完成事件对象的存储。

public void SaveEvent(IEvent idomainevent)
        {
            try
            {
                var domaineventobject = EventObject.FromDomainEvent(idomainevent);
                conn.Open();
                SqlParameter sqlparm = new SqlParameter("@AggregationRootId", System.Data.SqlDbType.UniqueIdentifier);
                sqlparm.Value = idomainevent.AggregationRootId;
                cmd =
                    new SqlCommand("select count(*) from DomainCommandAndEventObject where AggregationRootId=@AggregationRootId", conn);
                cmd.Parameters.Add(sqlparm);
                var count = cmd.ExecuteScalar();
                if(count!=null)
                {
                    domaineventobject.Version = int.Parse(count.ToString());
                }
                SqlParameter[] sqlparams = new SqlParameter[7];
                sqlparams[0] = new SqlParameter("@Id", System.Data.SqlDbType.UniqueIdentifier);
                sqlparams[0].Value = domaineventobject.Id;
                sqlparams[1] = new SqlParameter("@AggregationRootId", System.Data.SqlDbType.UniqueIdentifier);
                sqlparams[1].Value = domaineventobject.AggregationRootId;
                sqlparams[2] = new SqlParameter("@AssemblyQualifiedAggreateRooType", System.Data.SqlDbType.NVarChar);
                sqlparams[2].Value = domaineventobject.AssemblyQualifiedAggreateRooType;
                sqlparams[3] = new SqlParameter("@AssemblyQualifiedCommandAndEventType", System.Data.SqlDbType.NVarChar);
                sqlparams[3].Value = domaineventobject.AssemblyQualifiedCommandAndEventType;
                sqlparams[4] = new SqlParameter("@CreateDate", System.Data.SqlDbType.DateTime);
                sqlparams[4].Value = domaineventobject.CreateDate;
                sqlparams[5] = new SqlParameter("@Version", System.Data.SqlDbType.Int);
                sqlparams[5].Value = domaineventobject.Version;
                sqlparams[6] = new SqlParameter("@Data", System.Data.SqlDbType.VarBinary);
                sqlparams[6].Value = domaineventobject.Data;
                cmd = new SqlCommand("insert DomainCommandAndEventObject values
                (@Id,@AggregationRootId,@AssemblyQualifiedAggreateRooType,@AssemblyQualifiedCommandAndEventType,@CreateDate,@Version,@Data)", conn);
                foreach(var sqlparam in sqlparams)
                {
                    cmd.Parameters.Add(sqlparam);
                }
                cmd.ExecuteNonQuery();

            }
            catch(Exception error)
            {
                throw error;
            }
            finally
            {
                cmd.Dispose();
                conn.Close();
            }

这样，我们基本就实现了事件与存储方面的基础内容。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018-09-05 ，如有侵权请联系 cloudcommunity@tencent.com 删除

微服务