微服务实战(七):落地微服务架构到直销系统(实现命令与命令处理器)

我们先来看看CQRS架构,你对下图的架构还有印象吗?每个组件的功能都还清楚吗?如果有疑问,请查考文章《微服务实战(五):落地微服务架构到直销系统(构建高性能大并发系统)》。

前一篇文章已经实现了Event Store的基础功能部分,本篇文章我们通过C端的标准方式,实现一个下单的高并发命令端,来看看需要实现的具体流程:

1.前端用户调用一个下单Command WebApi,传递下单命令;下单Command WebApi接受到下单命令后,将下单命令数据投递到一个命令队列中,向前端用户返回一个信息。

2.下单Command Handler WebApi侦听命令队列中的下单命令,然后调用领域对象逻辑,将执行的结果也就是Order对象的当前状态持久化到Event Store中。

3.下单Command Handler WebApi将下单相关信息通过事件的方式发布到一个事件队列中。(用户事件处理器最终将订单信息更新到业务库中)

下面通过代码简单体现下过程:

1.定义创建订单命令:

 public class CreateOrderCommand:BaseEvent
    {
        public OrderDTO orderdto { get; set; }
        public CreateOrderCommand() { }
        public CreateOrderCommand(OrderDTO orderdto)
        {
            this.orderdto = orderdto;
            this.AggregationRootId = Guid.NewGuid();
            this.AssemblyQualifiedAggreateRooType = typeof(Orders).AssemblyQualifiedName;
            this.AssemblyQualifiedCommandAndEventType = this.GetType().AssemblyQualifiedName;
        }
    }

2.订单 Command WebApi接受前端用户传递的订单命令: 

[Produces("application/json")]
    [Route("api/Order")]
    public class OrderController : Controller
    {
        private readonly IEventBus commandbus;
        public OrderController(IEventBus commandbus)
        {
            this.commandbus = commandbus;
        }
        [HttpPost]
        [Route("CreateOrderCmd")]
        public ResultEntity<bool> CreateOrderCmd([FromBody] OrderDTO orderdto)
        {
            var result = new ResultEntity<bool>();
            try
            {
                var createordercommand = new CreateOrderCommand(orderdto);
                //发布命令到命令总线
                commandbus.Publish(createordercommand);
                result.IsSuccess = true;
                result.Msg = "下单处理中!";
            }
            catch(Exception error)
            {
                result.ErrorCode = 200;
                result.Msg = error.Message;
            }
            return result;
        }
    }

 当然需要定义要注入的命令总线:

public void ConfigureServices(IServiceCollection services)
        {
            services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_1);
            //定义要发布命令的命令总线
            services.AddSingleton<IEventHandlerExecutionContext>(new EventHandlerExecutionContext(services));
            var connectionFactory = new ConnectionFactory { HostName = "localhost" };
            services.AddSingleton<IEventBus>(sp => new RabbitMqEB(connectionFactory,
                sp.GetRequiredService<IEventHandlerExecutionContext>(), "exchange1", "direct", "ordercreatecommandqueue", 1));
        }

 3.订单Command Handler Web Api侦听订单命令,并订阅需要处理订单的订单命令处理器:

            //侦听订单创建命令队列里的消息
            services.AddSingleton<IEventHandlerExecutionContext>(new EventHandlerExecutionContext(services));
            var connectionFactory = new ConnectionFactory { HostName = "localhost" };
            services.AddSingleton<IEventBus>(sp => new RabbitMqEB(connectionFactory,
                sp.GetRequiredService<IEventHandlerExecutionContext>(), "exchange1", "direct", "ordercreatecommandqueue", 2));
            //订阅创建订单命令
            var commandbuss = app.ApplicationServices.GetServices<IEventBus>();
            var commandbus = commandbuss.ToList()[0];
            commandbus.Subscribe<CreateOrderCommand, OrderCreateCommandHandler>();

 4.实现订单命令处理器:

public class OrderCreateCommandHandler : IEventHandler
    {
        private readonly IServiceProvider iserviceprovider;
        public OrderCreateCommandHandler()
        {
            var iwebhost = FindIWebHost.GetWwebHost("OrderCommandHandler.WebApi");
            iserviceprovider = iwebhost.Services;

        }
        public Task<bool> HandleAsync<TEvent>(TEvent @event) where TEvent : IEvent
        {
            var orderdtocommand = @event as CreateOrderCommand;
            var orderdto = orderdtocommand.orderdto;
            var orderid = orderdtocommand.AggregationRootId;
            Orders order = new Orders();
            var productskus = new List<ProductSKU>();
            for (int i = 0; i < orderdto.ProductSPUNames.Count; i++)
            {
                var productsku = new ProductSKU();
                productsku.ProductSPUName = orderdto.ProductSPUNames[i];
                productsku.DealerPrice = orderdto.ProductDealerPrices[i];
                productsku.PV = orderdto.ProductPVS[i];
                productsku.Id = orderdto.ProductSKUIds[i];
                productsku.Spec = orderdto.ProductSepcs[i];
                productskus.Add(productsku);
            }
            var contact = new Contact();
            contact.ContactName = orderdto.ContactName;
            contact.ContactTel = orderdto.ContactTel;
            contact.Province = orderdto.Privence;
            contact.City = orderdto.City;
            contact.Zero = orderdto.Zero;
            contact.Street = orderdto.Street;
            //完成业务逻辑
            var orders = order.CreateOrders(orderid, orderdto.DealerId, productskus, orderdto.Counts,
                contact);

            var ordercreateevent = new OrderCreateEvent();
            ordercreateevent.AggregationRootId = orders.Id;
            ordercreateevent.AssemblyQualifiedAggreateRooType = orderdtocommand.AssemblyQualifiedAggreateRooType;
            ordercreateevent.AssemblyQualifiedCommandAndEventType = orderdtocommand.AssemblyQualifiedCommandAndEventType;
            ordercreateevent.CreateDate = orders.OrderDateTime;
            ordercreateevent.Id = orders.Id;
            ordercreateevent.OrderDateTime = orders.OrderDateTime;
            ordercreateevent.OrderDealerId = orders.OrderDealerId;
            ordercreateevent.OrderItems = orders.OrderItems;
            ordercreateevent.OrderStreet = orders.OrderStreet;
            ordercreateevent.OrderTotalPrice = orders.OrderTotalPrice;
            ordercreateevent.OrderTotalPV = orders.OrderTotalPV;
            ordercreateevent.Code = orders.Code;
            ordercreateevent.Telephone = orders.Telephone;
            ordercreateevent.Version = 0;
            //对创建订单事件持久化事件存储
            try
            {
                new DomainAndEventStorage().SaveEvent(ordercreateevent);
                var eventbuss = iserviceprovider.GetServices(typeof(IEventBus))
                    as IEnumerable<IEventBus>;
                var eventbusls = eventbuss.ToList();
                var eventbus = eventbusls[1];
                //发布到事件队列,用于未来持久化到业务库中
                eventbus.Publish(ordercreateevent);
            }
            catch(Exception error)
            {
                throw error;
            }
            return Task.FromResult(true);
        }
    }

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏丑胖侠

Zookeeper客户端API之创建会话(六)

Zookeeper对外提供了一套Java的客户端API。本篇博客主要讲一下创建会话。 创建项目 首选,创建一个基于maven管理的简单java工程。在pom文件...

21290
来自专栏dotnet & java

WCF 入门 (21)

其实不太了解为什么第21集才讲这个Binding,下面都是一些概念性的东西,不过作为一个入门视频,了解一下也无妨吧。

9350
来自专栏晓晨的专栏

asp.net core使用jexus部署在linux无法正确 获取远程ip的解决办法

20150
来自专栏菩提树下的杨过

[原创]WCF入门级使用教程(转载请注明出处)

开发环境:vs2008英文版(SP1) + IIS + Windows2003 整个解决方案有4个项目 01.WCF ---Class Libary项目,用于...

21770
来自专栏程序员的SOD蜜

“一切都是消息”--MSF(消息服务框架)之【请求-响应】模式(点对点)

在前一篇, “一切都是消息”--MSF(消息服务框架)入门简介, 我们介绍了MSF基于异步通信,支持请求-响应通信模式和发布-订阅通信模式,并且介绍了如何获取M...

46880
来自专栏抠抠空间

Django之admin的使用和源码剖析

19500
来自专栏码农阿宇

EF Core利用Transaction对数据进行回滚保护

What? 首先,说一下什么是EF Core中的Transaction Transaction允许以原子方式处理多个数据库操作,如果事务已提交,则所有操作都应用...

37850
来自专栏码农分享

定时从列表中爬今日通知信息,打包成windows服务

每天8点爬取今日发布的新闻和通知公告,将爬取后的信息保存到Excel文件中,将程序发布成windows服务,开机即可自动启动。

13020
来自专栏魏琼东

分享一个分布式消息总线,基于.NET Socket Tcp的发布-订阅框架,附代码下载

     在很多MIS项目之中都有这样的需求,需要一个及时、高效的的通知机制,即比如当使用者A完成了任务X,就需要立即告知使用者B任务X已经完成,在通常的情况下...

45800
来自专栏LanceToBigData

HttpClient(二)HttpClient使用Ip代理与处理连接超时

前言   其实前面写的那一点点东西都是轻轻点水,其实HttpClient还有很多强大的功能:   (1)实现了所有 HTTP 的方法(GET,POST,PUT,...

43380

扫码关注云+社区

领取腾讯云代金券