/// <summary>
          /// 交換器聲明
          /// </summary>
          /// <param name="iModel"></param>
          /// <param name="exchange">交換器</param>
          /// <param name="type">交換器類型：
          /// 1、Direct Exchange – 處理路由鍵。需要將一個(gè)隊(duì)列綁定到交換機(jī)上，要求該消息與一個(gè)特定的路由鍵完全
          /// 匹配。這是一個(gè)完整的匹配。如果一個(gè)隊(duì)列綁定到該交換機(jī)上要求路由鍵 “dog”，則只有被標(biāo)記為“dog”的
          /// 消息才被轉(zhuǎn)發(fā)，不會(huì)轉(zhuǎn)發(fā)dog.puppy，也不會(huì)轉(zhuǎn)發(fā)dog.guard，只會(huì)轉(zhuǎn)發(fā)dog
         /// 2、Fanout Exchange – 不處理路由鍵。你只需要簡(jiǎn)單的將隊(duì)列綁定到交換機(jī)上。一個(gè)發(fā)送到交換機(jī)的消息都
         /// 會(huì)被轉(zhuǎn)發(fā)到與該交換機(jī)綁定的所有隊(duì)列上。很像子網(wǎng)廣播，每臺(tái)子網(wǎng)內(nèi)的主機(jī)都獲得了一份復(fù)制的消息。Fanout
         /// 交換機(jī)轉(zhuǎn)發(fā)消息是最快的。
         /// 3、Topic Exchange – 將路由鍵和某模式進(jìn)行匹配。此時(shí)隊(duì)列需要綁定要一個(gè)模式上。符號(hào)“#”匹配一個(gè)或多
         /// 個(gè)詞，符號(hào)“*”匹配不多不少一個(gè)詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*”
         /// 只會(huì)匹配到“audit.irs”。</param>
         /// <param name="durable">持久化</param>
         /// <param name="autoDelete">自動(dòng)刪除</param>
         /// <param name="arguments">參數(shù)</param>
         private static void ExchangeDeclare(IModel iModel, string exchange, string type = ExchangeType.Direct,
             bool durable = true,
             bool autoDelete = false, IDictionary<string, object> arguments = null)
         {
             exchange = exchange.IsNullOrWhiteSpace() ? "" : exchange.Trim();
             iModel.ExchangeDeclare(exchange, type, durable, autoDelete, arguments);
         }
 
         /// <summary>
         /// 隊(duì)列聲明
         /// </summary>
         /// <param name="channel"></param>
         /// <param name="queue">隊(duì)列</param>
         /// <param name="durable">持久化</param>
         /// <param name="exclusive">排他隊(duì)列，如果一個(gè)隊(duì)列被聲明為排他隊(duì)列，該隊(duì)列僅對(duì)首次聲明它的連接可見，
         /// 并在連接斷開時(shí)自動(dòng)刪除。這里需要注意三點(diǎn)：其一，排他隊(duì)列是基于連接可見的，同一連接的不同信道是可
         /// 以同時(shí)訪問(wèn)同一個(gè)連接創(chuàng)建的排他隊(duì)列的。其二，“首次”，如果一個(gè)連接已經(jīng)聲明了一個(gè)排他隊(duì)列，其他連
         /// 接是不允許建立同名的排他隊(duì)列的，這個(gè)與普通隊(duì)列不同。其三，即使該隊(duì)列是持久化的，一旦連接關(guān)閉或者
         /// 客戶端退出，該排他隊(duì)列都會(huì)被自動(dòng)刪除的。這種隊(duì)列適用于只限于一個(gè)客戶端發(fā)送讀取消息的應(yīng)用場(chǎng)景。</param>
         /// <param name="autoDelete">自動(dòng)刪除</param>
         /// <param name="arguments">參數(shù)</param>
         private static void QueueDeclare(IModel channel, string queue, bool durable = true, bool exclusive = false,
            bool autoDelete = false, IDictionary<string, object> arguments = null)
         {
             queue = queue.IsNullOrWhiteSpace() ? "UndefinedQueueName" : queue.Trim();
             channel.QueueDeclare(queue, durable, exclusive, autoDelete, arguments);
        }
 
        /// <summary>
        /// 獲取Model
        /// </summary>
       /// <param name="exchange">交換機(jī)名稱</param>
        /// <param name="queue">隊(duì)列名稱</param>
         /// <param name="routingKey"></param>
         /// <param name="isProperties">是否持久化</param>
        /// <returns></returns>
         private static IModel GetModel(string exchange, string queue, string routingKey, bool isProperties = false)
        {
          return ModelDic.GetOrAdd(queue, key =>
            {
                 var model = _conn.CreateModel();
                ExchangeDeclare(model, exchange, ExchangeType.Fanout, isProperties);
                 QueueDeclare(model, queue, isProperties);
                model.QueueBind(queue, exchange, routingKey);
                ModelDic[queue] = model;
                 return model;
            });
        }
 
       /// <summary>
       /// 發(fā)布消息
         /// </summary>
       /// <param name="routingKey">路由鍵</param>
         /// <param name="body">隊(duì)列信息</param>
        /// <param name="exchange">交換機(jī)名稱</param>
        /// <param name="queue">隊(duì)列名</param>
       /// <param name="isProperties">是否持久化</param>
        /// <returns></returns>
         public void Publish(string exchange, string queue, string routingKey, string body, bool isProperties = false)
         {
           var channel = GetModel(exchange, queue, routingKey, isProperties);

             try
             {
               channel.BasicPublish(exchange, routingKey, null, body.SerializeUtf8());
             }
            catch (Exception ex)
            {
               throw ex.GetInnestException();
           }
       }        

/// <summary>
        /// 獲取Model
        /// </summary>
        /// <param name="queue">隊(duì)列名稱</param>
        /// <param name="isProperties"></param>
        /// <returns></returns>
        private static IModel GetModel(string queue, bool isProperties = false)
        {
            return ModelDic.GetOrAdd(queue, value =>
             {
                 var model = _conn.CreateModel();
                 QueueDeclare(model, queue, isProperties);

                 //每次消費(fèi)的消息數(shù)
                 model.BasicQos(0, 1, false);

                 ModelDic[queue] = model;

                 return model;
             });
        }    

        /// <summary>
        /// 接收消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="queue">隊(duì)列名稱</param>
        /// <param name="isProperties"></param>
        /// <param name="handler">消費(fèi)處理</param>
        /// <param name="isDeadLetter"></param>
        public void Subscribe<T>(string queue, bool isProperties, Action<T> handler, bool isDeadLetter) where T : class
        {
            //隊(duì)列聲明
            var channel = GetModel(queue, isProperties);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var msgStr = body.DeserializeUtf8();
                var msg = msgStr.FromJson<T>();
                try
                {
                    handler(msg);
                }
                catch (Exception ex)
                {
                    ex.GetInnestException().WriteToFile("隊(duì)列接收消息", "RabbitMq");
                    if (!isDeadLetter)
                        PublishToDead<DeadLetterQueue>(queue, msgStr, ex);
                }
                finally
                {
                    channel.BasicAck(ea.DeliveryTag, false);
                }
            };
            channel.BasicConsume(queue, false, consumer);
        }        

 /// <summary>
        /// 獲取消息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="exchange"></param>
        /// <param name="queue"></param>
        /// <param name="routingKey"></param>
        /// <param name="handler">消費(fèi)處理</param>
        private void Poll<T>(string exchange, string queue, string routingKey, Action<T> handler) where T : class
        {
            var channel = GetModel(exchange, queue, routingKey);

            var result = channel.BasicGet(queue, false);
            if (result.IsNull())
                return;

            var msg = result.Body.DeserializeUtf8().FromJson<T>();
            try
            {
                handler(msg);
            }
            catch (Exception ex)
            {
                ex.GetInnestException().WriteToFile("隊(duì)列接收消息", "RabbitMq");
            }
            finally
            {
                channel.BasicAck(result.DeliveryTag, false);
            }
        }    

 /// <summary>
        /// RPC客戶端
        /// </summary>
        /// <param name="exchange"></param>
        /// <param name="queue"></param>
        /// <param name="routingKey"></param>
        /// <param name="body"></param>
        /// <param name="isProperties"></param>
        /// <returns></returns>
        public string RpcClient(string exchange, string queue, string routingKey, string body, bool isProperties = false)
        {
            var channel = GetModel(exchange, queue, routingKey, isProperties);

            var consumer = new QueueingBasicConsumer(channel);
            channel.BasicConsume(queue, true, consumer);

            try
            {
                var correlationId = Guid.NewGuid().ToString();
                var basicProperties = channel.CreateBasicProperties();
                basicProperties.ReplyTo = queue;
                basicProperties.CorrelationId = correlationId;

                channel.BasicPublish(exchange, routingKey, basicProperties, body.SerializeUtf8());

                var sw = Stopwatch.StartNew();
                while (true)
                {
                    var ea = consumer.Queue.Dequeue();
                    if (ea.BasicProperties.CorrelationId == correlationId)
                    {
                        return ea.Body.DeserializeUtf8();
                    }

                    if (sw.ElapsedMilliseconds > 30000)
                        throw new Exception("等待響應(yīng)超時(shí)");
                }
            }
            catch (Exception ex)
            {
                throw ex.GetInnestException();
            }
        }    

        /// <summary>
        /// RPC服務(wù)端
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="exchange"></param>
        /// <param name="queue"></param>
        /// <param name="isProperties"></param>
        /// <param name="handler"></param>
        /// <param name="isDeadLetter"></param>
        public void RpcService<T>(string exchange, string queue, bool isProperties, Func<T, T> handler, bool isDeadLetter)
        {
            //隊(duì)列聲明
            var channel = GetModel(queue, isProperties);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var msgStr = body.DeserializeUtf8();
                var msg = msgStr.FromJson<T>();

                var props = ea.BasicProperties;
                var replyProps = channel.CreateBasicProperties();
                replyProps.CorrelationId = props.CorrelationId;

                try
                {
                    msg = handler(msg);
                }
                catch (Exception ex)
                {
                    ex.GetInnestException().WriteToFile("隊(duì)列接收消息", "RabbitMq");
                }
                finally
                {
                    channel.BasicPublish(exchange, props.ReplyTo, replyProps, msg.ToJson().SerializeUtf8());
                    channel.BasicAck(ea.DeliveryTag, false);
                }
            };
            channel.BasicConsume(queue, false, consumer);
        }