在這個微服務火熱的時代，如果不懂一點微服務相關的技術，想吹下牛都沒有法子，于是有必要了解學習一下，所以我最近看了下微服務相關的知識，
微服務所涉及的知識是很廣的，我這里只是講一下事件總線，當然，有現成很棒的框架如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:取消訂閱還沒有試過，我看了好多人寫的取消訂閱的方法是基于事件的思想，我也理解不了為啥，因為我覺得直接定義一個方法去實作就好了，
轉自 天天博客，歡迎訪問
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/1647.html
標籤：.NET Core
上一篇：授權權限服務設計決議