EventBus

Posted 2022-05-05 cloudsu

EventBus 是一种事件发布订阅模式,借助 EventBus 我们可以很好的实现组件之间，服务之间，系统之间的解耦以及相互通信的问题。

 

EventBus 相当于是定义一些抽象接口，可以用 MQ 来实现EventBus

1、模块的预处理模块，定义预处理方法，增加实现ILocalEventHandler，IDistributedEventHandler的服务增加到配置

LocalEventBusOptions、DistributedEventBusOptions，分别存储ITypeList<IEventHandler> Handlers get;

2、IEventHandle接口，分ILocalEventHandler<in TEvent>，IDistributedEventHandler

        /// <summary>
        /// Handler handles the event by implementing this method.
        /// </summary>
        /// <param name="eventData">Event data</param>
        Task HandleEventAsync(TEvent eventData);

IEventHandlerFactory,负责得到或创建事件处理器，三个实现IocEventHandlerFactory，SingleInstanceHandlerFactory，TransientEventHandlerFactory

 IEventHandlerDisposeWrapper GetHandler();


  public interface IEventHandlerDisposeWrapper : IDisposable
    
        IEventHandler EventHandler  get; 
    

  public class EventHandlerDisposeWrapper : IEventHandlerDisposeWrapper
    
        public IEventHandler EventHandler  get; 

        private readonly Action _disposeAction;

        public EventHandlerDisposeWrapper(IEventHandler eventHandler, Action disposeAction = null)
        
            _disposeAction = disposeAction;
            EventHandler = eventHandler;
        

        public void Dispose()
        
            _disposeAction?.Invoke();
        
    

 

3、EventBus：用来发布/订阅/取消订阅事件

public abstract Task PublishAsync(Type eventType, object eventData);

public abstract IDisposable Subscribe(Type eventType, IEventHandlerFactory factory);


 public abstract void Unsubscribe<TEvent>(Func<TEvent, Task> action) where TEvent : class;

 public abstract void Unsubscribe(Type eventType, IEventHandler handler);
 public abstract void Unsubscribe(Type eventType, IEventHandlerFactory factory);
public abstract void Unsubscribe(Type eventType, IEventHandlerFactory factory);
public abstract void UnsubscribeAll(Type eventType);

二、AbpRabbitMqModule

 [DependsOn(
        typeof(AbpJsonModule),
        typeof(AbpThreadingModule)
        )]
    public class AbpRabbitMqModule : AbpModule
    
        public override void ConfigureServices(ServiceConfigurationContext context)
        
            var configuration = context.Services.GetConfiguration();
            Configure<AbpRabbitMqOptions>(configuration.GetSection("RabbitMQ"));
        

        public override void OnApplicationShutdown(ApplicationShutdownContext context)
        
            context.ServiceProvider
                .GetRequiredService<IChannelPool>()
                .Dispose();

            context.ServiceProvider
                .GetRequiredService<IConnectionPool>()
                .Dispose();
        

三、AbpEventBusRabbitMqModule

public override Task PublishAsync(Type eventType, object eventData)
        
            var eventName = EventNameAttribute.GetNameOrDefault(eventType);
            var body = Serializer.Serialize(eventData);

            using (var channel = ConnectionPool.Get(RabbitMqEventBusOptions.ConnectionName).CreateModel())
            
                channel.ExchangeDeclare(
                    RabbitMqEventBusOptions.ExchangeName,
                    "direct",
                    durable: true
                );
                
                var properties = channel.CreateBasicProperties();
                properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent;

                channel.BasicPublish(
                   exchange: RabbitMqEventBusOptions.ExchangeName,
                    routingKey: eventName,
                    mandatory: true,
                    basicProperties: properties,
                    body: body
                );
            

            return Task.CompletedTask;
        

        public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
        
            var handlerFactories = GetOrCreateHandlerFactories(eventType);
            
            handlerFactories.Add(factory);

            if (handlerFactories.Count == 1) //TODO: Multi-threading!
            
                Consumer.BindAsync(EventNameAttribute.GetNameOrDefault(eventType));
            

            return new EventHandlerFactoryUnregistrar(this, eventType, factory);