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

在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();
            }

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

 

QQ讨论群：309287205 

微服务实战视频请关注微信公众号：