申请域名网站价格,网络营销服务的特点,郑州外贸网站建设哪家好,seo公司杭州很长一段时间以来#xff0c;我都在思考如何在ASP.NET Core的框架下#xff0c;实现一套完整的事件驱动型架构。这个问题看上去有点大#xff0c;其实主要目标是为了实现一个基于ASP.NET Core的微服务#xff0c;它能够非常简单地订阅来自于某个渠道的事件消息#xff0c;… 很长一段时间以来我都在思考如何在ASP.NET Core的框架下实现一套完整的事件驱动型架构。这个问题看上去有点大其实主要目标是为了实现一个基于ASP.NET Core的微服务它能够非常简单地订阅来自于某个渠道的事件消息并对接收到的消息进行处理于此同时它还能够向该渠道发送事件消息以便订阅该事件消息的消费者能够对消息数据做进一步处理。让我们回顾一下微服务之间通信的几种方式分为同步和异步两种。同步通信最常见的就是RESTful API而且非常简单轻量一个Request/Response回环就结束了异步通信最常见的就是通过消息渠道将载有特殊意义的数据的事件消息发送到消息渠道而对某种类型消息感兴趣的消费者就可以获取消息中所带信息并执行相应操作这也是我们比较熟知的事件驱动架构的一种表现形式。虽然事件驱动型架构看起来非常复杂从微服务的实现来看显得有些繁重但它的应用范围确实很广也为服务间通信提供了新的思路。了解DDD的朋友相信一定知道CQRS体系结构模式它就是一种事件驱动型架构。事实上实现一套完整的、安全的、稳定的、正确的事件驱动架构并不简单由于异步特性带来的一致性问题会非常棘手甚至需要借助一些基础结构层工具比如关系型数据库不错只能是关系型数据库来解决一些特殊问题。本文就打算带领大家一起探探路基于ASP.NET Core Web API实现一个相对比较简单的事件驱动架构然后引出一些有待深入思考的问题留在今后的文章中继续讨论。或许本文所引入的源代码无法直接用于生产环境但我希望本文介绍的内容能够给到读者一些启发并能够帮助解决实际中遇到的问题。术语约定本文会涉及一些相关的专业术语在此先作约定事件在某一特定时刻发生在某件事物上的一件事情例如在我撰写本文的时候电话铃响了消息承载事件数据的实体。事件的序列化/反序列化和传输都以消息的形式进行消息通信渠道一种带有消息路由功能的数据传输机制用以在消息的派发器和订阅器之间进行数据传输注意为了迎合描述的需要在下文中可能会混用事件和消息两个概念。一个简单的设计先从简单的设计开始基本上事件驱动型架构会有事件消息Events、事件订阅器Event Subscriber、事件派发器Event Publisher、事件处理器Event Handler以及事件总线Event Bus等主要组件它们之间的关系大致如下首先IEvent接口定义了事件消息更确切地说数据的基本结构几乎所有的事件都会有一个唯一标识符Id和一个事件发生的时间Timestamp这个时间通常使用UTC时间作为标准。IEventHandler定义了事件处理器接口显而易见它包含两个方法CanHandle方法用以确定传入的事件对象是否可被当前处理器所处理以及Handle方法它定义了事件的处理过程。IEvent和IEventHandler构成了事件处理的基本元素。然后就是IEventSubscriber与IEventPublisher接口。前者表示实现该接口的类型为事件订阅器它负责事件处理器的注册并侦听来自事件通信渠道上的消息一旦所获得的消息能够被某个处理器处理它就会指派该处理器对接收到的消息进行处理。因此IEventSubscriber会保持着对事件处理器的引用而对于实现了IEventPublisher接口的事件派发器而言它的主要任务就是将事件消息发送到消息通信渠道以便订阅端能够获得消息并进行处理。IEventBus接口表示消息通信渠道也就是大家所熟知的消息总线的概念。它不仅具有消息订阅的功能而且还具有消息派发的能力因此它会同时继承于IEventSubscriber和IEventPublisher接口。在上面的设计中通过接口分离消息总线的订阅器和派发器的角色是很有必要的因为两种角色的各自职责不一样这样的设计同时满足SOLID中的SRP和ISP两个准则。基于以上基础模型我们可以很快地将这个对象关系模型转换为C#代码public interface IEvent{ Guid Id { get; } DateTime Timestamp { get; }}public interface IEventHandler{ Taskbool HandleAsync(IEvent event, CancellationToken cancellationToken default); bool CanHandle(IEvent event);}public interface IEventHandlerin T : IEventHandler where T : IEvent{ Taskbool HandleAsync(T event, CancellationToken cancellationToken default);}public interface IEventPublisher : IDisposable{ Task PublishAsyncTEvent(TEvent event, CancellationToken cancellationToken default) where TEvent : IEvent;}public interface IEventSubscriber : IDisposable{ void Subscribe();}public interface IEventBus : IEventPublisher, IEventSubscriber { }短短30行代码就把我们的基本对象关系描述清楚了。对于上面的代码我们需要注意以下几点这段代码使用了C# 7.1的新特性default关键字Publish以及Handle方法被替换为支持异步调用的PublishAsync和HandleAsync方法它们会返回Task对象这样可以方便使用C#中async/await的编程模型由于我们的这个模型可以作为实现消息系统的通用模型并且会需要用到ASP.NET Core的项目中因此建议将这些接口的定义放在一个独立的NetStandard的Class Library中方便今后重用和扩展OK接口定义好了。实现呢下面我们实现一个非常简单的消息总线PassThroughEventBus。在今后的文章中我还会介绍如何基于RabbitMQ和Azure Service Bus实现不一样的消息总线。PassThroughEventBus顾名思义PassThroughEventBus表示当有消息被派发到消息总线时消息总线将不做任何处理与路由而是直接将消息推送到订阅方。在订阅方的事件监听函数中会通过已经注册的事件处理器对接收到的消息进行处理。整个过程并不会依赖于任何外部组件不需要引用额外的开发库只是利用现有的.NET数据结构来模拟消息的派发和订阅过程。因此PassThroughEventBus不具备容错和消息重发功能不具备消息存储和路由功能我们先实现这样一个简单的消息总线来体验事件驱动型架构的设计过程。我们可以使用.NET中的Queue或者ConcurrentQueue等基本数据结构来作为消息队列的实现与这些基本的数据结构相比消息队列本身有它自己的职责它需要在消息被推送进队列的同时通知调用方。当然PassThroughEventBus不需要依赖于Queue或者ConcurrentQueue它所要做的事情就是模拟一个消息队列当消息推送进来的时候立刻通知订阅方进行处理。同样为了分离职责我们可以引入一个EventQueue的实现如下从而将消息推送和路由的职责基础结构层的职责从消息总线中分离出来。?12345678910111213internal sealed class EventQueue{ public event System.EventHandlerEventProcessedEventArgs EventPushed; public EventQueue() { } public void Push(IEvent event) { OnMessagePushed(new EventProcessedEventArgs(event)); } private void OnMessagePushed(EventProcessedEventArgs e) this.EventPushed?.Invoke(this, e);}EventQueue中最主要的方法就是Push方法从上面的代码可以看到当EventQueue的Push方法被调用时它将立刻触发EventPushed事件它是一个.NET事件用以通知EventQueue对象的订阅者消息已经被派发。整个EventQueue的实现非常简单我们仅专注于事件的路由完全没有考虑任何额外的事情。接下来就是利用EventQueue来实现PassThroughEventBus。毫无悬念PassThroughEventBus需要实现IEventBus接口它的两个基本操作分别是Publish和Subscribe。在Publish方法中会将传入的事件消息转发到EventQueue上而Subscribe方法则会订阅EventQueue.EventPushed事件.NET事件而在EventPushed事件处理过程中会从所有已注册的事件处理器Event Handlers中找到能够处理所接收到的事件并对其进行处理。整个流程还是非常清晰的。以下便是PassThroughEventBus的实现代码?12345678910111213141516171819202122232425262728293031323334353637383940public sealed class PassThroughEventBus : IEventBus{ private readonly EventQueue eventQueue new EventQueue(); private readonly IEnumerableIEventHandler eventHandlers; public PassThroughEventBus(IEnumerableIEventHandler eventHandlers) { this.eventHandlers eventHandlers; } private void EventQueue_EventPushed(object sender, EventProcessedEventArgs e) (from eh in this.eventHandlers where eh.CanHandle(e.Event) select eh).ToList().ForEach(async eh await eh.HandleAsync(e.Event)); public Task PublishAsyncTEvent(TEvent event, CancellationToken cancellationToken default) where TEvent : IEvent Task.Factory.StartNew(() eventQueue.Push(event)); public void Subscribe() eventQueue.EventPushed EventQueue_EventPushed; #region IDisposable Support private bool disposedValue false; // To detect redundant calls void Dispose(bool disposing) { if (!disposedValue) { if (disposing) { this.eventQueue.EventPushed - EventQueue_EventPushed; } disposedValue true; } } public void Dispose() Dispose(true); #endregion}实现过程非常简单当然从这些代码也可以更清楚地了解到PassThroughEventBus不做任何路由处理更不会依赖于一个基础结构设施比如实现了AMQP的消息队列因此不要指望能够在生产环境中使用它。不过目前来看它对于我们接下来要讨论的事情还是会很有帮助的至少在我们引入基于RabbitMQ等实现的消息总线之前。同样地请将PassThroughEventBus实现在另一个NetStandard的Class Library中虽然它不需要额外的依赖但它毕竟是众多消息总线中的一种将它从接口定义的程序集中剥离开来好处有两点第一保证了定义接口的程序集的纯净度使得该程序集不需要依赖任何外部组件并确保了该程序集的职责单一性即为消息系统的实现提供基础类库第二将PassThroughEventBus置于独立的程序集中有利于调用方针对IEventBus进行技术选择比如如果开发者选择使用基于RabbitMQ的实现那么只需要引用基于RabbitMQ实现IEventBus接口的程序集就可以了而无需引用包含了PassThroughEventBus的程序集。这一点我觉得可以归纳为框架设计中“隔离依赖关系Dependency Segregation”的准则。好了基本组件都定义好了接下来让我们一起基于ASP.NET Core Web API来做一个RESTful服务并接入上面的消息总线机制实现消息的派发和订阅。Customer RESTful API我们仍然以客户管理的RESTful API为例子不过我们不会过多地讨论如何去实现管理客户信息的RESTful服务那并不是本文的重点。作为一个案例我使用ASP.NET Core 2.0 Web API建立了这个服务使用Visual Studio 2017 15.5做开发并在CustomersController中使用Dapper来对客户信息CRUD。后台基于SQL Server 2017 Express Edition使用SQL Server Management Studio能够让我方便地查看数据库操作的结果。RESTful API的实现假设我们的客户信息只包含客户ID和名称下面的CustomersController代码展示了我们的RESTful服务是如何保存并读取客户信息的。当然我已经将本文的代码通过Github开源开源协议为MIT虽然商业友好但毕竟是案例代码没有经过测试所以请谨慎使用。本文源代码的使用我会在文末介绍。?1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950[Route(api/[controller])]public class CustomersController : Controller{ private readonly IConfiguration configuration; private readonly string connectionString; public CustomersController(IConfiguration configuration) { this.configuration configuration; this.connectionString configuration[mssql:connectionString]; } // 获取指定ID的客户信息 [HttpGet({id})] public async TaskIActionResult Get(Guid id) { const string sql SELECT [CustomerId] AS Id, [CustomerName] AS Name FROM [dbo].[Customers] WHERE [CustomerId]id; using (var connection new SqlConnection(connectionString)) { var customer await connection.QueryFirstOrDefaultAsyncModel.Customer(sql, new { id }); if (customer null) { return NotFound(); } return Ok(customer); } } // 创建新的客户信息 [HttpPost] public async TaskIActionResult Create([FromBody] dynamic model) { var name (string)model.Name; if (string.IsNullOrEmpty(name)) { return BadRequest(); } const string sql INSERT INTO [dbo].[Customers] ([CustomerId], [CustomerName]) VALUES (Id, Name); using (var connection new SqlConnection(connectionString)) { var customer new Model.Customer(name); await connection.ExecuteAsync(sql, customer); return Created(Url.Action(Get, new { id customer.Id }), customer.Id); } }}代码一如既往的简单Web API控制器通过Dapper简单地实现了客户信息的创建和返回。我们不妨测试一下使用下面的Invoke-RestMethod PowerShell指令发送Post请求通过上面的Create方法创建一个用户可以看到response中已经返回了新建客户的ID号。接下来继续使用Invoke-RestMethod来获取新建客户的详细信息OKAPI调试完全没有问题。下面我们将这个案例再扩充一下我们希望这个API在完成客户信息创建的同时向外界发送一条“客户信息已创建”的事件并设置一个事件处理器负责将该事件的详细内容保存到数据库中。加入事件总线和消息处理机制首先我们在ASP.NET Core Web API项目上添加对以上两个程序集的引用然后按常规做法在ConfigureServices方法中将PassThroughEventBus添加到IoC容器中?12345public void ConfigureServices(IServiceCollection services){ services.AddMvc(); services.AddSingletonIEventBus, PassThroughEventBus();}在此将事件总线注册为单例Singleton服务是因为它不保存状态。理论上讲使用单例服务时需要特别注意服务实例对象的生命周期管理因为它的生命周期是整个应用程序级别在程序运行的过程中由其引用的对象资源将无法释放因此当程序结束运行时需要合理地将这些资源dispose掉。好在ASP.NET Core的依赖注入框架中已经帮我们处理过了因此对于上面的PassThroughEventBus单例注册我们不需要过多担心程序执行结束并正常退出时依赖注入框架会自动帮我们dispose掉PassThroughEventBus的单例实例。那么对于单例实例来说我们是否只需要通过AddSingleton方法进行注册就可以了而无需关注它是否真的被dispose了呢答案是否定的有兴趣的读者可以参考微软的官方文档在下一篇文章中我会对这部分内容做些介绍。接下来我们需要定义一个CustomerCreatedEvent对象表示“客户信息已经创建”这一事件信息同时再定义一个CustomerCreatedEventHandler事件处理器用来处理从PassThroughEventBus接收到的事件消息。代码如下当然也很简单?1234567891011121314151617181920212223242526272829public class CustomerCreatedEvent : IEvent{ public CustomerCreatedEvent(string customerName) { this.Id Guid.NewGuid(); this.Timestamp DateTime.UtcNow; this.CustomerName customerName; } public Guid Id { get; } public DateTime Timestamp { get; } public string CustomerName { get; }}public class CustomerCreatedEventHandler : IEventHandlerCustomerCreatedEvent{ public bool CanHandle(IEvent event) event.GetType().Equals(typeof(CustomerCreatedEvent)); public Taskbool HandleAsync(CustomerCreatedEvent event, CancellationToken cancellationToken default) { return Task.FromResult(true); public Taskbool HandleAsync(IEvent event, CancellationToken cancellationToken default) CanHandle(event) ? HandleAsync((CustomerCreatedEvent)event, cancellationToken) : Task.FromResult(false);}两者分别实现了我们最开始定义好的IEvent和IEventHandler接口。在CustomerCreatedEventHandler类的第一个HandleAsync重载方法中我们暂且让它简单地返回一个true值表示事件处理成功。下面要做的事情就是在客户信息创建成功后向事件总线发送CustomerCreatedEvent事件以及在ASP.NET Core Web API程序启动的时候注册CustomerCreatedEventHandler实例并调用事件总线的Subscribe方法使其开始侦听事件的派发行为。于是CustomerController需要依赖IEventBus并且在CustomerController.Create方法中需要通过调用IEventBus的Publish方法将事件发送出去。现对CustomerController的实现做一些调整调整后代码如下?123456789101112131415161718192021222324252627282930313233343536373839[Route(api/[controller])]public class CustomersController : Controller{ private readonly IConfiguration configuration; private readonly string connectionString; private readonly IEventBus eventBus; public CustomersController(IConfiguration configuration, IEventBus eventBus) { this.configuration configuration; this.connectionString configuration[mssql:connectionString]; this.eventBus eventBus; } // 创建新的客户信息 [HttpPost] public async TaskIActionResult Create([FromBody] dynamic model) { var name (string)model.Name; if (string.IsNullOrEmpty(name)) { return BadRequest(); } const string sql INSERT INTO [dbo].[Customers] ([CustomerId], [CustomerName]) VALUES (Id, Name); using (var connection new SqlConnection(connectionString)) { var customer new Model.Customer(name); await connection.ExecuteAsync(sql, customer); await this.eventBus.PublishAsync(new CustomerCreatedEvent(name)); return Created(Url.Action(Get, new { id customer.Id }), customer.Id); } } // Get方法暂且省略}然后修改Startup.cs中的ConfigureServices方法将CustomerCreatedEventHandler注册进来?1234567public void ConfigureServices(IServiceCollection services){ services.AddMvc(); services.AddTransientIEventHandler, CustomerCreatedEventHandler(); services.AddSingletonIEventBus, PassThroughEventBus();}并且调用Subscribe方法开始侦听消息总线?123456789101112public void Configure(IApplicationBuilder app, IHostingEnvironment env){ var eventBus app.ApplicationServices.GetRequiredServiceIEventBus(); eventBus.Subscribe(); if (env.IsDevelopment()) { app.UseDeveloperExceptionPage(); } app.UseMvc();}OK现在让我们在CustomerCreatedEventHandler的HandleAsync方法上设置个断点按下F5启用Visual Studio 2017调试然后重新使用Invoke-RestMethod命令发送一个Post请求可以看到HandleAsync方法上的断点被命中同时事件已被正确派发数据库中的数据也被正确更新目前还差最后一小步就是在HandleAsync中将CustomerCreatedEvent对象的数据序列化并保存到数据库中。当然这也不难同样可以考虑使用Dapper或者直接使用ADO.NET甚至使用比较重量级的Entity Framework Core都可以实现。那就在此将这个问题留给感兴趣的读者朋友自己搞定啦。小结到这里基本上本文的内容也就告一段落了回顾一下本文一开始就提出了一种相对简单的消息系统和事件驱动型架构的设计模型并实现了一个最简单的事件总线PassThroughEventBus。随后结合一个实际的ASP.NET Core Web API案例了解了在RESTful API中实现事件消息派发和订阅的过程并实现了在事件处理器中对获得的事件消息进行处理。然而我们还有很多问题需要更深入地思考比如如果事件处理器需要依赖基础结构层组件依赖关系如何管理组件生命周期如何管理如何实现基于RabbitMQ或者Azure Service Bus的事件总线如果在数据库更新成功后事件发送失败怎么办如何保证事件处理的顺序等等。。。在接下来的文章中我会尽力做更详细的介绍。源代码的使用本系列文章的源代码在https://github.com/daxnet/edasample这个Github Repo里通过不同的release tag来区分针对不同章节的源代码。本文的源代码请参考chapter_1这个tag如下接下来还将会有chapter_2、chapter_3等这些tag对应本系列文章的第二部分、第三部分等等。敬请期待。原文地址:http://www.cnblogs.com/daxnet/p/8082694.html.NET社区新闻深度好文欢迎访问公众号文章汇总 http://www.csharpkit.com
