事件总线EventBus

作用

为什么要使用EventBus
Eventbus可以帮助不同的微服务之间实现解耦，使它们能够独立开发、部署和扩展。微服务可以通过发布和订阅事件的方式进行通信，而不需要显式地了解其他微服务的存在和实现细节。
MQ的eventBus底层使用了事务来保持数据的一致性，而内存总线使用UOW工作单元实现数据一致性
Eventbus使用发布/订阅模型，其中组件可以发布事件或订阅感兴趣的事件，当事件被发布到Eventbus时，Eventbus会负责将事件传播给所有订阅者；

内存事件总线和MQ事件总线的区别
内存事件总线是一种在应用程序内部实现的事件传输机制，用于在应用程序的不同组件之间进行通信和数据传递。
相比于使用消息队列等外部服务，内存事件总线可以提供更低的延迟和更高的吞吐量。
但MQ事件总线可以适用于不同的应用，适用于微服务架构之间的信息传递。

框架EventBus使用方法

先通过DI注入

  /// <summary>
  /// 事件总线
  /// </summary>
  private readonly IEventBus _eventBus;
  /// <summary>
  /// 构造
  /// </summary>
  public TestManager(IServiceProvider serviceProvider,  IEventBus eventBus) : base(serviceProvider)
  {
      _repository = repository ?? throw new ArgumentNullException(nameof(repository));
      _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
  }

发布订阅消息

 public async Task ValidateDeleteAsync(List<Test> entities)
 {
     await _eventBus.PublishAsync(new TestDeleteValidationEvent(entities.Select(t => t.Id).ToList()));
 }

创建订阅类

  /// <summary>
  /// 删除验证事件
  /// </summary>
  public record TestDeleteValidationEvent : IEvent
  {
      public IEnumerable<Guid> CodeIds { get; }
      /// <summary>
      /// 初始化缺陷代码删除验证事件
      /// </summary>
      public TestCodeDeleteValidationEvent(IEnumerable<Guid> CodeIds)
      {
          CodeIds = CodeIds;
      }
  }

接受订阅消息

public class TestDeleteValidationEventHandles : EventHandlerBase<TestDeleteValidationEvent >
{
    private readonly ITestCodeProcessRepository _repository;
    public TestCodeDeleteValidationEventHandles(ITestCodeProcessRepository repository)
    {
        _repository = repository;
    }
    public override async Task HandleAsync(TestCodeDeleteValidationEvent @event, CancellationToken cancellationToken)
    {
        if (@event.CodeIds != null)
        {
            if (await _repository.ExistsAsync(t => @event.TestCodeIds.Contains(t.TestId)))
            {
                throw new Warning(Util.Localization.Helper.Format("DeleteHandleWarning"));
            }
        }
    }
}

实现原理

为保证数据的一致性（强一致性）
当我们要发出事件时，我们会把事件的存储逻辑与我们的业务逻辑的事务合并，在同一个事务里提交，只要事件存储的逻辑或者业务代码写入逻辑有一个失败就回滚到最初状态。
同理，订阅逻辑也相同。

配置发布对象以及订阅响应

接下来，我们需要创建一个事件和一个事件处理程序。事件是我们希望在系统中传递的消息，而事件处理程序则负责处理这些事件。

创建一个名为 MyEvent 的类，用于表示一个简单的事件：

public class MyEvent
{
    public string Message { get; set; }
}

接下来，我们创建一个名为 MyEventHandler 的类，用于处理 MyEvent 事件：

public class MyEventHandler
{
    public void Handle(MyEvent myEvent)
    {
        // 处理事件的逻辑
    }

配置 RabbitMQ 连接

我们需要在应用程序中配置 RabbitMQ 连接。在你的项目中创建一个名为 RabbitMQConfig 的类，用于存储连接配置信息：

public class RabbitMQConfig
{
    public string Hostname { get; set; }
    public string Username { get; set; }
    public string Password { get; set; }
}

在你的应用程序的配置文件（例如 appsettings.json）中添加 RabbitMQ 的连接配置：

{
  "RabbitMQConfig": {
    "Hostname": "localhost",
    "Username": "guest",
    "Password": "guest"
  }
}

消息发布实现

消息发布的实现采用的是将发送信息序列化，发布给订阅后再由订阅反序列化。

using RabbitMQ.Client;
using System.Text;
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json;
// ...
var factory = new ConnectionFactory
{
    HostName = configuration.GetSection("RabbitMQConfig:Hostname").Value,
    UserName = configuration.GetSection("RabbitMQConfig:Username").Value,
    Password = configuration.GetSection("RabbitMQConfig:Password").Value
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{   
    //定义mq管道
    channel.QueueDeclare(queue: "my_queue",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);
    var myEvent = new MyEvent { Message = "Hello, world!" };
    var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(myEvent));
    channel.BasicPublish(exchange: "",
                         routingKey: "my_queue",
                         basicProperties: null,
                         body: body);
}

消息订阅实现

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using Newtonsoft.Json;
// ...
var factory = new ConnectionFactory
{
    HostName = configuration.GetSection("RabbitMQConfig:Hostname").Value,
    UserName = configuration.GetSection("RabbitMQConfig:Username").Value,
    Password = configuration.GetSection("RabbitMQConfig:Password").Value
};
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    //定义mq管道
    channel.QueueDeclare(queue: "my_queue",
                         durable: false,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);
    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        var myEvent = JsonConvert.DeserializeObject<MyEvent>(message);
        // 处理接收到的事件对象
        Console.WriteLine("Received event: " + myEvent.Message);
    };
    channel.BasicConsume(queue: "my_queue",
                         autoAck: true,
                         consumer: consumer);
    // 保持程序运行，等待事件对象的到达
    Console.WriteLine("Waiting for events...");
    Console.ReadLine();
}

目录