Yuebon.EventBus.RabbitMQ/EventBusRabbitMQ.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

namespace Yuebon.EventBus.RabbitMQ;
/// <summary>
/// 基于RabbitMQ的事件总线
/// </summary>
public class EventBusRabbitMQ : IEventBus, IDisposable
{
    const string BROKER_NAME = "yuebon_event_bus";
    const string AUTOFAC_SCOPE_NAME = "yuebon_event_bus";

    private readonly IRabbitMQPersistentConnection _persistentConnection;
    private readonly ILogger<EventBusRabbitMQ> _logger;
    private readonly IEventBusSubscriptionsManager _subsManager;
    private readonly ILifetimeScope _autofac;
    private readonly int _retryCount;

    private IModel _consumerChannel;
    private string _queueName;

    /// <summary>
    /// RabbitMQ事件总线
    /// </summary>
    /// <param name="persistentConnection">RabbitMQ持久连接</param>
    /// <param name="logger">日志</param>
    /// <param name="autofac">autofac容器</param>
    /// <param name="subsManager">事件总线订阅管理器</param>
    /// <param name="queueName">队列名称</param>
    /// <param name="retryCount">重试次数</param>
    public EventBusRabbitMQ(IRabbitMQPersistentConnection persistentConnection, ILogger<EventBusRabbitMQ> logger,
            ILifetimeScope autofac, IEventBusSubscriptionsManager subsManager, string queueName = null, int retryCount = 5)
    {
        _persistentConnection = persistentConnection ?? throw new ArgumentNullException(nameof(persistentConnection));
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
        _subsManager = subsManager ?? new InMemoryEventBusSubscriptionsManager();
        _queueName = queueName;
        _consumerChannel = CreateConsumerChannel();
        _autofac = autofac;
        _retryCount = retryCount;
        _subsManager.OnEventRemoved += SubsManager_OnEventRemoved;
    }

    /// <summary>
    /// 订阅管理器事件
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="eventName"></param>
    private void SubsManager_OnEventRemoved(object sender, string eventName)
    {
        if (!_persistentConnection.IsConnected)
        {
            _persistentConnection.TryConnect();
        }

        using var channel = _persistentConnection.CreateModel();
        channel.QueueUnbind(queue: _queueName,
            exchange: BROKER_NAME,
            routingKey: eventName);

        if (_subsManager.IsEmpty)