在这个微服务火热的时代，如果不懂一点微服务相关的技术，想吹下牛都没有法子。于是有必要了解学习一下。所以我最近看了下微服务相关的知识。
微服务所涉及的知识是很广的，我这里只是讲一下事件总线，当然，有现成很棒的框架如CAP，但是我这里只是为了去体验去更深入的了解事件总线，去了解它的工作流程，所以我自己写了一个基于RabbitMQ的事件总线。
1，运行rabbitmq；
2，创建解决方案，模拟分布式如下图（我这里之前学下了一下微服务的网关，所以会有Gateway，所以要运行3个程序，并且运行Consul做服务发现）：
3，实现api1的发布功能：
1，创建IEventBus作为抽象接口，实际上你可以用多个MQ来实现它，我这里只是使用RabbitMQ，所以建一个EventBusRabbitMQ来实现接口
public interface IEventBus { } public class EventBusRabbitMQ : IEventBus { }
2，然后新建一个类，用来实现事件总线的DI注入：
serviceDescriptors.AddTransient<IEventBus, EventBusRabbitMQ>();
3，发布消息，为了能够让不同的服务都这个消息类型，并且可以使其作为参数传递，所以我们需要一个基类作为消息的总线：EventData，然后我们服务定义的每一个消息类型都必须继承这个类：
public class CreateUserEvent : EventData { public string Name { get; set; } public string Address { get; set; } public DateTime CreateTime { get; set; } }
既然有消息那就会有消息事件，还需要一个EventHandler来驱动，所以每个服务的消息对象都应该有一个事件驱动的类，当然这个驱动类应该在订阅方，应为发布方只负责发布消息，至于消息的处理事件则应该由订阅方实现，下面再讲。其中消息的发布是基于rabbitmq，网上有很多实现的例子，这里只是我的写法，以EventData作为参数：
public void Publish(EventData eventData) { string routeKey = eventData.GetType().Name; channel.QueueDeclare(queueName, true, false, false, null); string message = JsonConvert.SerializeObject(eventData); byte[] body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchangeName, routeKey, null, body); }
然后访问apil来模拟消息发布：
[HttpGet] [Route("api/user/eventbus")] public IActionResult Eventbus() { CreateUserEvent user = new CreateUserEvent(); user.Name = "hanh"; user.Address = "hubei"; user.CreateTime = DateTime.Now; _eventBus.Publish(user); return Ok("ok"); }
4，实现api2的订阅功能
刚已经将了订阅应该会实现消息的事件处理，那么就会有UserEventHandler，继承EventHandler，来处理消息：
public class UserEventHandler : IEventHandler<CreateUserEvent>, IEventHandler<UpdateUserEvent> { private readonly ILogger<UserEventHandler> _logger; public UserEventHandler(ILogger<UserEventHandler> logger) { _logger = logger; } public async Task Handler(CreateUserEvent eventData) { _logger.LogInformation(JsonConvert.SerializeObject(eventData)); await Task.FromResult(0); } public async Task Handler(UpdateUserEvent eventData) { await Task.FromResult(0); } }
然后会开始处理订阅，大致的思路就是根据EventData作为key，然后每个EventData都应该有一个泛型的EventHandler<>接口，然后将其作为value存入内存中，同时rabbitmq绑定消息队列，当消息到达时，自动处理消息事件，获取到发布消息的类型名字，然后我们根据类型名字从内从中获取到它的EventData的类型，接着再根据这个类型，通过.net core内置的IOC来获取到它的实现类，每个EventData的类型会匹配到不同的EventHandler，所以会完成CRUD。至此，大致的订阅已经实现了：
public void AddSub<T, TH>() where T : EventData where TH : IEventHandler { Type eventDataType = typeof(T); Type handlerType = typeof(TH); if (!eventhandlers.ContainsKey(typeof(T))) eventhandlers.TryAdd(eventDataType, handlerType); if(!eventTypes.ContainsKey(eventDataType.Name)) eventTypes.TryAdd(eventDataType.Name, eventDataType); if (assemblyTypes != null) { Type implementationType = assemblyTypes.FirstOrDefault(s => handlerType.IsAssignableFrom(s)); if (implementationType == null) throw new ArgumentNullException("未找到{0}的实现类", handlerType.FullName); _serviceDescriptors.AddTransient(handlerType, implementationType); } } public void Subscribe<T, TH>() where T : EventData where TH : IEventHandler { _eventBusManager.AddSub<T, TH>(); channel.QueueBind(queueName, exchangeName, typeof(T).Name); channel.QueueDeclare(queueName, true, false, false, null); var consumer = new EventingBasicConsumer(channel); consumer.Received +=async (model, ea) => { string eventName = ea.RoutingKey; byte[] resp = ea.Body.ToArray(); string body = Encoding.UTF8.GetString(resp); _log.LogInformation(body); try { Type eventType = _eventBusManager.FindEventType(eventName); T eventData = (T)JsonConvert.DeserializeObject(body, eventType); IEventHandler<T> eventHandler = _eventBusManager.FindHandlerType(eventType) as IEventHandler<T>; await eventHandler.Handler(eventData); } catch (Exception ex) { throw ex; } }; channel.BasicConsume(queueName, true, consumer); }
5，测试，访问api1，发布成功，然后api2会同时打印出信息：
最后给大家贴出核心代码，如果想看完整的请访问地址 https://github.com/Hansdas/Micro
using Micro.Core.EventBus.RabbitMQ;using System;using System.Collections.Generic;using System.Text;namespace Micro.Core.EventBus{ public interface IEventBus { /// <summary> /// 发布 /// </summary> /// <param name="eventData"></param> void Publish(EventData eventData); /// <summary> /// 订阅 /// </summary> /// <typeparam name="T"></typeparam> /// <typeparam name="TH"></typeparam> void Subscribe<T, TH>() where T : EventData where TH : IEventHandler; /// <summary> /// 取消订阅 /// </summary> /// <typeparam name="T"></typeparam> /// <typeparam name="TH"></typeparam> void Unsubscribe<T, TH>() where T : EventData where TH : IEventHandler; }}
using log4net;using Micro.Core.EventBus.RabbitMQ.IImplementation;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.DependencyInjection.Extensions;using Microsoft.Extensions.Logging;using Newtonsoft.Json;using RabbitMQ.Client;using RabbitMQ.Client.Events;using System;using System.Collections.Generic;using System.Reflection;using System.Text;namespace Micro.Core.EventBus.RabbitMQ{ public class EventBusRabbitMQ : IEventBus { /// <summary> /// 队列名称 /// </summary> private string queueName = "QUEUE"; /// <summary> /// 交换机名称 /// </summary> private string exchangeName = "directName"; /// <summary> /// 交换类型 /// </summary> private string exchangeType = "direct"; private IFactoryRabbitMQ _factory; private IEventBusManager _eventBusManager; private ILogger<EventBusRabbitMQ> _log; private readonly IConnection connection; private readonly IModel channel; public EventBusRabbitMQ(IFactoryRabbitMQ factory, IEventBusManager eventBusManager, ILogger<EventBusRabbitMQ> log) { _factory = factory; _eventBusManager = eventBusManager; _eventBusManager.OnRemoveEventHandler += OnRemoveEvent; _log = log; connection = _factory.CreateConnection(); channel = connection.CreateModel(); } private void OnRemoveEvent(object sender, ValueTuple<Type, Type> args) { channel.QueueUnbind(queueName, exchangeName, args.Item1.Name); } public void Publish(EventData eventData) { string routeKey = eventData.GetType().Name; channel.QueueDeclare(queueName, true, false, false, null); string message = JsonConvert.SerializeObject(eventData); byte[] body = Encoding.UTF8.GetBytes(message); channel.BasicPublish(exchangeName, routeKey, null, body); } public void Subscribe<T, TH>() where T : EventData where TH : IEventHandler { _eventBusManager.AddSub<T, TH>(); channel.QueueBind(queueName, exchangeName, typeof(T).Name); channel.QueueDeclare(queueName, true, false, false, null); var consumer = new EventingBasicConsumer(channel); consumer.Received +=async (model, ea) => { string eventName = ea.RoutingKey; byte[] resp = ea.Body.ToArray(); string body = Encoding.UTF8.GetString(resp); _log.LogInformation(body); try { Type eventType = _eventBusManager.FindEventType(eventName); T eventData = (T)JsonConvert.DeserializeObject(body, eventType); IEventHandler<T> eventHandler = _eventBusManager.FindHandlerType(eventType) as IEventHandler<T>; await eventHandler.Handler(eventData); } catch (Exception ex) { throw ex; } }; channel.BasicConsume(queueName, true, consumer); } public void Unsubscribe<T, TH>() where T : EventData where TH : IEventHandler { if (_eventBusManager.HaveAddHandler(typeof(T))) { _eventBusManager.RemoveEventSub<T, TH>(); } } }}
using System;using System.Collections.Generic;using System.Text;namespace Micro.Core.EventBus{ public interface IEventBusManager { /// <summary> /// 取消订阅事件 /// </summary> event EventHandler<ValueTuple<Type, Type>> OnRemoveEventHandler; /// <summary> /// 订阅 /// </summary> /// <typeparam name="T"></typeparam> /// <typeparam name="TH"></typeparam> void AddSub<T, TH>() where T : EventData where TH : IEventHandler; /// <summary> /// 取消订阅 /// </summary> /// <typeparam name="T"></typeparam> /// <typeparam name="TH"></typeparam> void RemoveEventSub<T, TH>() where T : EventData where TH : IEventHandler; /// <summary> /// 是否包含实体类型 /// </summary> /// <param name="type"></param> /// <returns></returns> bool HaveAddHandler(Type eventDataType); /// <summary> /// 根据实体名称寻找类型 /// </summary> /// <param name="eventName"></param> /// <returns></returns> Type FindEventType(string eventName); /// <summary> /// 根据实体类型寻找它的领域事件驱动 /// </summary> /// <param name="eventDataType"></param> /// <returns></returns> object FindHandlerType(Type eventDataType); }}
using Micro.Core.Configure;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.DependencyInjection.Extensions;using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Linq;using System.Reflection;using System.Text;namespace Micro.Core.EventBus{ internal class EventBusManager : IEventBusManager { public event EventHandler<ValueTuple<Type, Type>> OnRemoveEventHandler; private static ConcurrentDictionary<Type, Type> eventhandlers=new ConcurrentDictionary<Type, Type>(); private readonly ConcurrentDictionary<string, Type> eventTypes = new ConcurrentDictionary<string, Type>(); private readonly IList<Type> assemblyTypes; private readonly IServiceCollection _serviceDescriptors; private Func<IServiceCollection, IServiceProvider> _buildServiceProvider; public EventBusManager(IServiceCollection serviceDescriptors,Func<IServiceCollection,IServiceProvider> buildServiceProvicer) { _serviceDescriptors = serviceDescriptors; _buildServiceProvider = buildServiceProvicer; string dllName = ConfigurationProvider.configuration.GetSection("EventHandler.DLL").Value; if (!string.IsNullOrEmpty(dllName)) { assemblyTypes = Assembly.Load(dllName).GetTypes(); } } private void OnRemoveEvent(Type eventDataType, Type handler) { if (OnRemoveEventHandler != null) { OnRemoveEventHandler(this, new ValueTuple<Type, Type>(eventDataType, handler)); } } public void AddSub<T, TH>() where T : EventData where TH : IEventHandler { Type eventDataType = typeof(T); Type handlerType = typeof(TH); if (!eventhandlers.ContainsKey(typeof(T))) eventhandlers.TryAdd(eventDataType, handlerType); if(!eventTypes.ContainsKey(eventDataType.Name)) eventTypes.TryAdd(eventDataType.Name, eventDataType); if (assemblyTypes != null) { Type implementationType = assemblyTypes.FirstOrDefault(s => handlerType.IsAssignableFrom(s)); if (implementationType == null) throw new ArgumentNullException("未找到{0}的实现类", handlerType.FullName); _serviceDescriptors.AddTransient(handlerType, implementationType); } } public void RemoveEventSub<T, TH>() where T : EventData where TH : IEventHandler { OnRemoveEvent(typeof(T), typeof(TH)); } public bool HaveAddHandler(Type eventDataType) { if (eventhandlers.ContainsKey(eventDataType)) return true; return false; } public Type FindEventType(string eventName) { if(!eventTypes.ContainsKey(eventName)) throw new ArgumentException(string.Format("eventTypes不存在类名{0}的key", eventName)); return eventTypes[eventName]; } public object FindHandlerType(Type eventDataType) { if(!eventhandlers.ContainsKey(eventDataType)) throw new ArgumentException(string.Format("eventhandlers不存在类型{0}的key", eventDataType.FullName)); var obj = _buildServiceProvider(_serviceDescriptors).GetService(eventhandlers[eventDataType]); if (eventhandlers[eventDataType].IsAssignableFrom(obj.GetType())) return obj; return null; } }}
using Micro.Core.Configure; using Micro.Core.EventBus.RabbitMQ;using Micro.Core.EventBus.RabbitMQ.IImplementation;using Microsoft.AspNetCore.Builder;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.DependencyInjection.Extensions;using System;using System.Collections.Generic;using System.Linq;using System.Reflection;using System.Text;namespace Micro.Core.EventBus{ public static class EventBusBuilder { public static EventBusOption eventBusOption; public static IServiceCollection AddEventBus(this IServiceCollection serviceDescriptors) { eventBusOption= ConfigurationProvider.GetModel<EventBusOption>("EventBusOption"); switch (eventBusOption.MQProvider) { case MQProvider.RabbitMQ: serviceDescriptors.AddTransient<IEventBus, EventBusRabbitMQ>(); serviceDescriptors.AddTransient(typeof(IFactoryRabbitMQ), factiory => { return new FactoryRabbitMQ(eventBusOption); }); break; } EventBusManager eventBusManager = new EventBusManager(serviceDescriptors,s=>s.BuildServiceProvider()); serviceDescriptors.AddSingleton<IEventBusManager>(eventBusManager); return serviceDescriptors; } }}api1
public class Startup { public Startup(IConfiguration configuration) { Configuration = configuration; } public IConfiguration Configuration { get; } // This method gets called by the runtime. Use this method to add services to the container. public void ConfigureServices(IServiceCollection services) { services.AddControllers(); services.AddEventBus(); } }
api2:
using System;using System.Collections.Generic;using System.Linq;using System.Threading.Tasks;using Micro.Core.Configure;using Micro.Core.Consul;using Micro.Core.EventBus;using Micro.Services.Domain;using Microsoft.AspNetCore.Builder;using Microsoft.AspNetCore.Hosting;using Microsoft.AspNetCore.HttpsPolicy;using Microsoft.AspNetCore.Mvc;using Microsoft.Extensions.Configuration;using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.Hosting;using Microsoft.Extensions.Logging;using ConfigurationProvider = Micro.Core.Configure.ConfigurationProvider;namespace WebApi3{ public class Startup { public Startup(IConfiguration configuration) { Configuration = configuration; } public IConfiguration Configuration { get; } // This method gets called by the runtime. Use this method to add services to the container. public void ConfigureServices(IServiceCollection services) { services.AddControllers(); services.AddEventBus(); } // This method gets called by the runtime. Use this method to configure the HTTP request pipeline. public void Configure(IApplicationBuilder app, IWebHostEnvironment env) { var eventBus= app.ApplicationServices.GetRequiredService<IEventBus>(); eventBus.Subscribe<CreateUserEvent, IEventHandler<CreateUserEvent>>(); eventBus.Subscribe<UpdateUserEvent, IEventHandler<UpdateUserEvent>>(); if (env.IsDevelopment()) { app.UseDeveloperExceptionPage(); } } }}
ps:取消订阅还没有试过，我看了好多人写的取消订阅的方法是基于事件的思想，我也理解不了为啥，因为我觉得直接定义一个方法去实现就好了。
转自 天天博客，欢迎访问
基于rabbitmq的事件总线net a porter、 派代、 海拍客、 Offer to Buyers、 中东电商平台SOUQ的基本知识大全！ 、 亚马逊怎么发货到FBA仓库？亚马逊FBA发货注意事项、 沙巴岛 沙巴岛旅游 沙巴岛景点 沙巴岛旅游攻略实用篇、 香港旅游攻略 去香港旅游指南全新版、 国内第一仙山 武当山旅游全攻略、
No comments:
Post a Comment