Yuebon.NetCore/Yuebon.Commons/EventBus/RabbitMQPersistent/EventBusRabbitMQ.cs

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

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

using Autofac;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Polly;
using Polly.Retry;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System.Net.Sockets;
using System.Text;
using System.Text.Json;
using Yuebon.Commons.EventBus;
using Yuebon.Commons.EventBus.Abstractions;
using Yuebon.Commons.EventBus.Events;
using Yuebon.Commons.EventBus.Extensions;

namespace Yuebon.Commons.EventBus.RabbitMQPersistent;
/// <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>