基於rabbitmq的事件總線

時間 2020-07-21

原文原文鏈接

在這個微服務火熱的時代，若是不懂一點微服務相關的技術，想吹下牛都沒有法子。因而有必要了解學習一下。因此我最近看了下微服務相關的知識。
微服務所涉及的知識是很廣的，我這裏只是講一下事件總線，固然，有現成很棒的框架如CAP，可是我這裏只是爲了去體驗去更深刻的瞭解事件總線，去了解它的工做流程，因此我本身寫了一個基於RabbitMQ的事件總線。
1，運行rabbitmq；
2，建立解決方案，模擬分佈式以下圖（我這裏以前學下了一下微服務的網關，因此會有Gateway，因此要運行3個程序，而且運行Consul作服務發現）：

3，實現api1的發佈功能：
1，建立IEventBus做爲抽象接口，實際上你能夠用多個MQ來實現它，我這裏只是使用RabbitMQ，因此建一個EventBusRabbitMQ來實現接口html

 public interface IEventBus
    {
    }
   public class EventBusRabbitMQ : IEventBus
    {
    }

2，而後新建一個類，用來實現事件總線的DI注入：git

serviceDescriptors.AddTransient<IEventBus, EventBusRabbitMQ>();

3，發佈消息，爲了可以讓不一樣的服務都這個消息類型，而且能夠使其做爲參數傳遞，因此咱們須要一個基類做爲消息的總線：EventData，而後咱們服務定義的每個消息類型都必須繼承這個類：github

 public class CreateUserEvent : EventData
    {
        public string Name { get; set; }
        public string Address { get; set; }
        public DateTime CreateTime { get; set; }

    }

既然有消息那就會有消息事件，還須要一個EventHandler來驅動，因此每一個服務的消息對象都應該有一個事件驅動的類，固然這個驅動類應該在訂閱方，應爲發佈方只負責發佈消息，至於消息的處理事件則應該由訂閱方實現，下面再講。其中消息的發佈是基於rabbitmq，網上有不少實現的例子，這裏只是個人寫法，以EventData做爲參數：api

 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來模擬消息發佈：app

 [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。至此，大體的訂閱已經實現了：async

  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;
        }

    }
}

api1ide

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

        }
    }
}