Yi.Framework.Net6/Yi.Framework.Core/RabbitMQInvoker.cs

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

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

using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Yi.Framework.Common.IOCOptions;

namespace Yi.Framework.Core
{
    /// <summary>
    /// 一个Exchange----多个Queue-----弄个缓存映射关系，初始化+支持全新绑定
    /// 全局单例使用
    /// 
    /// 关系应该是直接配置到RabbitMQ了---程序只是向某个位置写入即可
    /// 
    /// 
    /// 全量更新--耗时---阻塞实时更新---换不同的exchange？
    /// </summary>
    public class RabbitMQInvoker
    {
        #region Identity
        private readonly RabbitMQOptions _rabbitMQOptions;
        private readonly string _HostName = null;
        private readonly string _UserName = null;
        private readonly string _Password = null;
        private readonly int _Port = 0;
        public RabbitMQInvoker(IOptionsMonitor<RabbitMQOptions> optionsMonitor) : this(optionsMonitor.CurrentValue.HostName, optionsMonitor.CurrentValue.UserName, optionsMonitor.CurrentValue.Password,optionsMonitor.CurrentValue.Port)
        {
            this._rabbitMQOptions = optionsMonitor.CurrentValue;
        }

        public RabbitMQInvoker(string hostName, string userName = "cc", string password = "cc",int port= 5672)
        {
            this._HostName = hostName;
            this._UserName = userName;
            this._Password = password;
            this._Port = port;
        }
        #endregion

        #region Init
        private static object RabbitMQInvoker_BindQueueLock = new object();
        private static Dictionary<string, bool> RabbitMQInvoker_ExchangeQueue = new Dictionary<string, bool>();
        private void InitBindQueue(RabbitMQConsumerModel rabbitMQConsumerModel)
        {
            if (!RabbitMQInvoker_ExchangeQueue.ContainsKey($"InitBindQueue_{rabbitMQConsumerModel.ExchangeName}_{rabbitMQConsumerModel.QueueName}"))
            {
                lock (RabbitMQInvoker_BindQueueLock)
                {
                    if (!RabbitMQInvoker_ExchangeQueue.ContainsKey($"InitBindQueue_{rabbitMQConsumerModel.ExchangeName}_{rabbitMQConsumerModel.QueueName}"))
                    {
                        this.InitConnection();
                        using (IModel channel = _CurrentConnection.CreateModel())
                        {
                            channel.ExchangeDeclare(exchange: rabbitMQConsumerModel.ExchangeName, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
                            channel.QueueDeclare(queue: rabbitMQConsumerModel.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
                            channel.QueueBind(queue: rabbitMQConsumerModel.QueueName, exchange: rabbitMQConsumerModel.ExchangeName, routingKey: string.Empty, arguments: null);