欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

基于RabbitMQ.Client组件实现RabbitMQ可复用的 ConnectionPool(连接池)

程序员文章站 2022-03-06 11:28:26
一、本文产生原由: 之前文章《总结消息队列RabbitMQ的基本用法》已对RabbitMQ的安装、用法都做了详细说明,而本文主要是针对在高并发且单次从RabbitMQ中消费消息时,出现了连接数不足、连接响应较慢、RabbitMQ服务器崩溃等各种性能问题的解方案,之所以会出现我列举的这些问题,究基根源 ......

一、本文产生原由:  

 之前文章《总结消息队列RabbitMQ的基本用法》已对RabbitMQ的安装、用法都做了详细说明,而本文主要是针对在高并发且单次从RabbitMQ中消费消息时,出现了连接数不足、连接响应较慢、RabbitMQ服务器崩溃等各种性能问题的解方案,之所以会出现我列举的这些问题,究基根源,其实是TCP连接创建与断开太过频繁所致,这与我们使用ADO.NET来访问常规的关系型DB(如:SQL SERVER、MYSQL)有所不同,在访问DB时,我们一般都建议大家使用using包裹,目的是每次创建完DB连接,使用完成后自动释放连接,避免不必要的连接数及资源占用。可能有人会问,为何访问DB,可以每次创建再断开连接,都没有问题,而同样访问MQ(本文所指的MQ均是RabbitMQ),每次创建再断开连接,如果在高并发且创建与断开频率高的时候,会出现性能问题呢?其实如果了解了DB的连接创建与断开以及MQ的连接创建与断开原理就知道其中的区别了。这里我简要说明一下,DB连接与MQ连接 其实底层都是基于TCP连接,创建TCP连接肯定是有资源消耗的,是非常昂贵的,原则上尽可能少的去创建与断开TCP连接,DB创建连接、MQ创建连接可以说是一样的,但在断开销毁连接上就有很大的不同,DB创建连接再断开时,默认情况下是把该连接回收到连接池中,下次如果再有DB连接创建请求,则先判断DB连接池中是否有空闲的连接,若有则直接复用,若没有才创建连接,这样就达到了TCP连接的复用,而MQ创建连接都是新创建的TCP连接,断开时则直接断开TCP连接,简单粗暴,看似资源清理更彻底,但若在高并发高频率每次都重新创建与断开MQ连接,则性能只会越来越差(上面说过TCP连接是非常昂贵的),我在公司项目中就出现了该问题,后面在技术总监的指导下,对MQ的连接创建与断开作了优化,实现了类似DB连接池的概念。

连接池,故名思义,连接的池子,所有的连接作为一种资源集中存放在池中,需要使用时就可以到池中获取空闲连接资源,用完后再放回池中,以此达到连接资源的有效重用,同时也控制了资源的过度消耗与浪费(资源多少取决于池子的容量)

二、源代码奉献(可直接复制应用到大家的项目中) 

下面就先贴出实现MQHelper(含连接池)的源代码:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using RabbitMQ.Util;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Web.Caching;
using System.Web;
using System.Configuration;
using System.IO;
using System.Collections.Concurrent;
using System.Threading;
using System.Runtime.CompilerServices;

namespace Zuowj.Core
{
    public class MQHelper
    {
        private const string CacheKey_MQConnectionSetting = "MQConnectionSetting";
        private const string CacheKey_MQMaxConnectionCount = "MQMaxConnectionCount";

        private readonly static ConcurrentQueue<IConnection> FreeConnectionQueue;//空闲连接对象队列
        private readonly static ConcurrentDictionary<IConnection, bool> BusyConnectionDic;//使用中(忙)连接对象集合
        private readonly static ConcurrentDictionary<IConnection, int> MQConnectionPoolUsingDicNew;//连接池使用率
        private readonly static Semaphore MQConnectionPoolSemaphore;
        private readonly static object freeConnLock = new object(), addConnLock = new object();
        private static int connCount = 0;

        public const int DefaultMaxConnectionCount = 30;//默认最大保持可用连接数
        public const int DefaultMaxConnectionUsingCount = 10000;//默认最大连接可访问次数


        private static int MaxConnectionCount
        {
            get
            {
                if (HttpRuntime.Cache[CacheKey_MQMaxConnectionCount] != null)
                {
                    return Convert.ToInt32(HttpRuntime.Cache[CacheKey_MQMaxConnectionCount]);
                }
                else
                {
                    int mqMaxConnectionCount = 0;
                    string mqMaxConnectionCountStr = ConfigurationManager.AppSettings[CacheKey_MQMaxConnectionCount];
                    if (!int.TryParse(mqMaxConnectionCountStr, out mqMaxConnectionCount) || mqMaxConnectionCount <= 0)
                    {
                        mqMaxConnectionCount = DefaultMaxConnectionCount;
                    }

                    string appConfigPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "App.config");
                    HttpRuntime.Cache.Insert(CacheKey_MQMaxConnectionCount, mqMaxConnectionCount, new CacheDependency(appConfigPath));

                    return mqMaxConnectionCount;
                }

            }
        }

        /// <summary>
        /// 建立连接
        /// </summary>
        /// <param name="hostName">服务器地址</param>
        /// <param name="userName">登录账号</param>
        /// <param name="passWord">登录密码</param>
        /// <returns></returns>
        private static ConnectionFactory CrateFactory()
        {
            var mqConnectionSetting = GetMQConnectionSetting();
            var connectionfactory = new ConnectionFactory();
            connectionfactory.HostName = mqConnectionSetting[0];
            connectionfactory.UserName = mqConnectionSetting[1];
            connectionfactory.Password = mqConnectionSetting[2];
            if (mqConnectionSetting.Length > 3) //增加端口号
            {
                connectionfactory.Port = Convert.ToInt32(mqConnectionSetting[3]);
            }
            return connectionfactory;
        }

        private static string[] GetMQConnectionSetting()
        {
            string[] mqConnectionSetting = null;
            if (HttpRuntime.Cache[CacheKey_MQConnectionSetting] == null)
            {
                //MQConnectionSetting=Host IP|;userid;|;password
                string mqConnSettingStr = ConfigurationManager.AppSettings[CacheKey_MQConnectionSetting];
                if (!string.IsNullOrWhiteSpace(mqConnSettingStr))
                {
                    mqConnSettingStr = EncryptUtility.Decrypt(mqConnSettingStr);//解密MQ连接字符串,若项目中无此需求可移除,EncryptUtility是一个AES的加解密工具类,大家网上可自行查找
                    if (mqConnSettingStr.Contains(";|;"))
                    {
                        mqConnectionSetting = mqConnSettingStr.Split(new[] { ";|;" }, StringSplitOptions.RemoveEmptyEntries);
                    }
                }

                if (mqConnectionSetting == null || mqConnectionSetting.Length < 3)
                {
                    throw new Exception("MQConnectionSetting未配置或配置不正确");
                }

                string appConfigPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "App.config");
                HttpRuntime.Cache.Insert(CacheKey_MQConnectionSetting, mqConnectionSetting, new CacheDependency(appConfigPath));
            }
            else
            {
                mqConnectionSetting = HttpRuntime.Cache[CacheKey_MQConnectionSetting] as string[];
            }

            return mqConnectionSetting;
        }




        public static IConnection CreateMQConnection()
        {
            var factory = CrateFactory();
            factory.AutomaticRecoveryEnabled = true;//自动重连
            var connection = factory.CreateConnection();
            connection.AutoClose = false;
            return connection;
        }


        static MQHelper()
        {
            FreeConnectionQueue = new ConcurrentQueue<IConnection>();
            BusyConnectionDic = new ConcurrentDictionary<IConnection, bool>();
            MQConnectionPoolUsingDicNew = new ConcurrentDictionary<IConnection, int>();//连接池使用率
            MQConnectionPoolSemaphore = new Semaphore(MaxConnectionCount, MaxConnectionCount, "MQConnectionPoolSemaphore");//信号量,控制同时并发可用线程数

        }

        public static IConnection CreateMQConnectionInPoolNew()
        {

        SelectMQConnectionLine:

            MQConnectionPoolSemaphore.WaitOne();//当<MaxConnectionCount时,会直接进入,否则会等待直到空闲连接出现

            IConnection mqConnection = null;
            if (FreeConnectionQueue.Count + BusyConnectionDic.Count < MaxConnectionCount)//如果已有连接数小于最大可用连接数,则直接创建新连接
            {
                lock (addConnLock)
                {
                    if (FreeConnectionQueue.Count + BusyConnectionDic.Count < MaxConnectionCount)
                    {
                        mqConnection = CreateMQConnection();
                        BusyConnectionDic[mqConnection] = true;//加入到忙连接集合中
                        MQConnectionPoolUsingDicNew[mqConnection] = 1;
                        //  BaseUtil.Logger.DebugFormat("Create a MQConnection:{0},FreeConnectionCount:{1}, BusyConnectionCount:{2}", mqConnection.GetHashCode().ToString(), FreeConnectionQueue.Count, BusyConnectionDic.Count);
                        return mqConnection;
                    }
                }
            }


            if (!FreeConnectionQueue.TryDequeue(out mqConnection)) //如果没有可用空闲连接,则重新进入等待排队
            {
                // BaseUtil.Logger.DebugFormat("no FreeConnection,FreeConnectionCount:{0}, BusyConnectionCount:{1}", FreeConnectionQueue.Count, BusyConnectionDic.Count);
                goto SelectMQConnectionLine;
            }
            else if (MQConnectionPoolUsingDicNew[mqConnection] + 1 > DefaultMaxConnectionUsingCount || !mqConnection.IsOpen) //如果取到空闲连接,判断是否使用次数是否超过最大限制,超过则释放连接并重新创建
            {
                mqConnection.Close();
                mqConnection.Dispose();
                // BaseUtil.Logger.DebugFormat("close > DefaultMaxConnectionUsingCount mqConnection,FreeConnectionCount:{0}, BusyConnectionCount:{1}", FreeConnectionQueue.Count, BusyConnectionDic.Count);

                mqConnection = CreateMQConnection();
                MQConnectionPoolUsingDicNew[mqConnection] = 0;
                // BaseUtil.Logger.DebugFormat("create new mqConnection,FreeConnectionCount:{0}, BusyConnectionCount:{1}", FreeConnectionQueue.Count, BusyConnectionDic.Count);
            }

            BusyConnectionDic[mqConnection] = true;//加入到忙连接集合中
            MQConnectionPoolUsingDicNew[mqConnection] = MQConnectionPoolUsingDicNew[mqConnection] + 1;//使用次数加1

            // BaseUtil.Logger.DebugFormat("set BusyConnectionDic:{0},FreeConnectionCount:{1}, BusyConnectionCount:{2}", mqConnection.GetHashCode().ToString(), FreeConnectionQueue.Count, BusyConnectionDic.Count);

            return mqConnection;
        }

        private static void ResetMQConnectionToFree(IConnection connection)
        {
            lock (freeConnLock)
            {
                bool result = false;
                if (BusyConnectionDic.TryRemove(connection, out result)) //从忙队列中取出
                {
                    //  BaseUtil.Logger.DebugFormat("set FreeConnectionQueue:{0},FreeConnectionCount:{1}, BusyConnectionCount:{2}", connection.GetHashCode().ToString(), FreeConnectionQueue.Count, BusyConnectionDic.Count);
                }
                else
                {
                    // BaseUtil.Logger.DebugFormat("failed TryRemove BusyConnectionDic:{0},FreeConnectionCount:{1}, BusyConnectionCount:{2}", connection.GetHashCode().ToString(), FreeConnectionQueue.Count, BusyConnectionDic.Count);
                }

                if (FreeConnectionQueue.Count + BusyConnectionDic.Count > MaxConnectionCount)//如果因为高并发出现极少概率的>MaxConnectionCount,则直接释放该连接
                {
                    connection.Close();
                    connection.Dispose();
                }
                else
                {
                    FreeConnectionQueue.Enqueue(connection);//加入到空闲队列,以便持续提供连接服务
                }

                MQConnectionPoolSemaphore.Release();//释放一个空闲连接信号

                //Interlocked.Decrement(ref connCount);
                //BaseUtil.Logger.DebugFormat("Enqueue FreeConnectionQueue:{0},FreeConnectionCount:{1}, BusyConnectionCount:{2},thread count:{3}", connection.GetHashCode().ToString(), FreeConnectionQueue.Count, BusyConnectionDic.Count,connCount);
            }
        }


        /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="connection">消息队列连接对象</param>
        /// <typeparam name="T">消息类型</typeparam>
        /// <param name="queueName">队列名称</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="msg">消息</param>
        /// <returns></returns>
        public static string SendMsg(IConnection connection, string queueName, string msg, bool durable = true)
        {
            try
            {

                using (var channel = connection.CreateModel())//建立通讯信道
                {
                    // 参数从前面开始分别意思为:队列名称,是否持久化,独占的队列,不使用时是否自动删除,其他参数
                    channel.QueueDeclare(queueName, durable, false, false, null);

                    var properties = channel.CreateBasicProperties();
                    properties.DeliveryMode = 2;//1表示不持久,2.表示持久化

                    if (!durable)
                        properties = null;

                    var body = Encoding.UTF8.GetBytes(msg);
                    channel.BasicPublish("", queueName, properties, body);
                }


                return string.Empty;
            }
            catch (Exception ex)
            {
                return ex.ToString();
            }
            finally
            {
                ResetMQConnectionToFree(connection);
            }
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        /// <param name="connection">消息队列连接对象</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="durable">是否持久化</param>
        /// <param name="dealMessage">消息处理函数</param>
        /// <param name="saveLog">保存日志方法,可选</param>
        public static void ConsumeMsg(IConnection connection, string queueName, bool durable, Func<string, ConsumeAction> dealMessage, Action<string, Exception> saveLog = null)
        {
            try
            {

                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(queueName, durable, false, false, null); //获取队列 
                    channel.BasicQos(0, 1, false); //分发机制为触发式

                    var consumer = new QueueingBasicConsumer(channel); //建立消费者
                    // 从左到右参数意思分别是:队列名称、是否读取消息后直接删除消息,消费者
                    channel.BasicConsume(queueName, false, consumer);

                    while (true)  //如果队列中有消息
                    {
                        ConsumeAction consumeResult = ConsumeAction.RETRY;
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); //获取消息
                        string message = null;

                        try
                        {
                            var body = ea.Body;
                            message = Encoding.UTF8.GetString(body);
                            consumeResult = dealMessage(message);
                        }
                        catch (Exception ex)
                        {
                            if (saveLog != null)
                            {
                                saveLog(message, ex);
                            }
                        }
                        if (consumeResult == ConsumeAction.ACCEPT)
                        {
                            channel.BasicAck(ea.DeliveryTag, false);  //消息从队列中删除
                        }
                        else if (consumeResult == ConsumeAction.RETRY)
                        {
                            channel.BasicNack(ea.DeliveryTag, false, true); //消息重回队列
                        }
                        else
                        {
                            channel.BasicNack(ea.DeliveryTag, false, false); //消息直接丢弃
                        }
                    }
                }

            }
            catch (Exception ex)
            {
                if (saveLog != null)
                {
                    saveLog("QueueName:" + queueName, ex);
                }

                throw ex;
            }
            finally
            {
                ResetMQConnectionToFree(connection);
            }
        }


        /// <summary>
        /// 依次获取单个消息
        /// </summary>
        /// <param name="connection">消息队列连接对象</param>
        /// <param name="QueueName">队列名称</param>
        /// <param name="durable">持久化</param>
        /// <param name="dealMessage">处理消息委托</param>
        public static void ConsumeMsgSingle(IConnection connection, string QueueName, bool durable, Func<string, ConsumeAction> dealMessage)
        {
            try
            {

                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(QueueName, durable, false, false, null); //获取队列 
                    channel.BasicQos(0, 1, false); //分发机制为触发式

                    uint msgCount = channel.MessageCount(QueueName);

                    if (msgCount > 0)
                    {
                        var consumer = new QueueingBasicConsumer(channel); //建立消费者
                        // 从左到右参数意思分别是:队列名称、是否读取消息后直接删除消息,消费者
                        channel.BasicConsume(QueueName, false, consumer);

                        ConsumeAction consumeResult = ConsumeAction.RETRY;
                        var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); //获取消息
                        try
                        {
                            var body = ea.Body;
                            var message = Encoding.UTF8.GetString(body);
                            consumeResult = dealMessage(message);
                        }
                        catch (Exception ex)
                        {
                            throw ex;
                        }
                        finally
                        {
                            if (consumeResult == ConsumeAction.ACCEPT)
                            {
                                channel.BasicAck(ea.DeliveryTag, false);  //消息从队列中删除
                            }
                            else if (consumeResult == ConsumeAction.RETRY)
                            {
                                channel.BasicNack(ea.DeliveryTag, false, true); //消息重回队列
                            }
                            else
                            {
                                channel.BasicNack(ea.DeliveryTag, false, false); //消息直接丢弃
                            }
                        }
                    }
                    else
                    {
                        dealMessage(string.Empty);
                    }
                }

            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ResetMQConnectionToFree(connection);
            }
        }


        /// <summary>
        /// 获取队列消息数
        /// </summary>
        /// <param name="connection"></param>
        /// <param name="QueueName"></param>
        /// <returns></returns>
        public static int GetMessageCount(IConnection connection, string QueueName)
        {
            int msgCount = 0;
            try
            {

                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare(QueueName, true, false, false, null); //获取队列 
                    msgCount = (int)channel.MessageCount(QueueName);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ResetMQConnectionToFree(connection);
            }

            return msgCount;
        }


    }

    public enum ConsumeAction
    {
        ACCEPT,  // 消费成功
        RETRY,   // 消费失败,可以放回队列重新消费
        REJECT,  // 消费失败,直接丢弃
    }
}

现在对上述代码的核心点作一个简要的说明:

先说一下静态构造函数:

FreeConnectionQueue 用于存放空闲连接对象队列,为何使用Queue,因为当我从中取出1个空闲连接后,空闲连接数就应该少1个,这个Queue很好满足这个需求,而且这个Queue是并发安全的Queue哦(ConcurrentQueue)

BusyConnectionDic 忙(使用中)连接对象集合,为何这里使用字典对象呢,因为当我用完后,需要能够快速的找出使用中的连接对象,并能快速移出,同时重新放入到空闲队列FreeConnectionQueue ,达到连接复用

MQConnectionPoolUsingDicNew 连接使用次数记录集合,这个只是辅助记录连接使用次数,以便可以计算一个连接的已使用次数,当达到最大使用次数时,则应断开重新创建

MQConnectionPoolSemaphore 这个是信号量,这是控制并发连接的重要手段,连接池的容量等同于这个信号量的最大可并行数,保证同时使用的连接数不超过连接池的容量,若超过则会等待;

具体步骤说明:

1.MaxConnectionCount:最大保持可用连接数(可以理解为连接池的容量),可以通过CONFIG配置,默认为30; 

2.DefaultMaxConnectionUsingCount:默认最大连接可访问次数,我这里没有使用配置,而是直接使用常量固定为1000,大家若有需要可以改成从CONFIG配置,参考MaxConnectionCount的属性设置(采取了依赖缓存)

3.CreateMQConnectionInPoolNew:从连接池中创建MQ连接对象,这个是核心方法,是实现连接池的地方,代码中已注释了重要的步骤逻辑,这里说一下实现思路:

  3.1 通过MQConnectionPoolSemaphore.WaitOne() 利用信号量的并行等待方法,如果当前并发超过信号量的最大并行度(也就是作为连接池的最大容量),则需要等待空闲连接池,防止连接数超过池的容量,如果并发没有超过池的容量,则可以进入获取连接的逻辑;

  3.2FreeConnectionQueue.Count + BusyConnectionDic.Count < MaxConnectionCount,如果空闲连接队列+忙连接集合的总数小于连接池的容量,则可以直接创建新的MQ连接,否则FreeConnectionQueue.TryDequeue(out mqConnection) 尝试从空闲连接队列中获取一个可用的空闲连接使用,若空闲连接都没有,则需要返回到方法首行,重新等待空闲连接;

  3.3MQConnectionPoolUsingDicNew[mqConnection] + 1 > DefaultMaxConnectionUsingCount || !mqConnection.IsOpen 如果取到空闲连接,则先判断使用次数是否超过最大限制,超过则释放连接或空闲连接已断开连接也需要重新创建,否则该连接可用;

  3.4BusyConnectionDic[mqConnection] = true;加入到忙连接集合中,MQConnectionPoolUsingDicNew[mqConnection] = MQConnectionPoolUsingDicNew[mqConnection] + 1; 使用次数加1,确保每使用一次连接,连接次数能记录

4.ResetMQConnectionToFree:重置释放连接对象,这个是保证MQ连接用完后能够回收到空闲连接队列中(即:回到连接池中),而不是直接断开连接,这个方法很简单就不作作过多说明。

 好了,都说明了如何实现含连接池的MQHelper,现在再来举几个例子来说明如何用:

三、实际应用(简单易上手)

获取并消费一个消息:

        public string GetMessage(string queueName)
        {
            string message = null;
            try
            {
                var connection = MQHelper.CreateMQConnectionInPoolNew();

                MQHelper.ConsumeMsgSingle(connection, queueName, true, (msg) =>
                {
                    message = msg;
                    return ConsumeAction.ACCEPT;
                });
            }
            catch (Exception ex)
            {
                BaseUtil.Logger.Error(string.Format("MQHelper.ConsumeMsgSingle Error:{0}", ex.Message), ex);
                message = "ERROR:" + ex.Message;
            }

            //BaseUtil.Logger.InfoFormat("第{0}次请求,从消息队列(队列名称:{1})中获取消息值为:{2}", Interlocked.Increment(ref requestCount), queueName, message);


            return message;


        }

 发送一个消息:

        public string SendMessage(string queueName, string msg)
        {
            string result = null;
            try
            {
                var connection = MQHelper.CreateMQConnectionInPoolNew();

                result = MQHelper.SendMsg(connection, queueName, msg);
            }
            catch (Exception ex)
            {
                BaseUtil.Logger.Error(string.Format("MQHelper.SendMessage Error:{0}", ex.Message), ex);
                result = ex.Message;
            }

            return result;
        }

 获取消息队列消息数:

        public int GetMessageCount(string queueName)
        {
            int result = -1;
            try
            {
                var connection = MQHelper.CreateMQConnectionInPoolNew();

                result = MQHelper.GetMessageCount(connection, queueName);
            }
            catch (Exception ex)
            {
                BaseUtil.Logger.Error(string.Format("MQHelper.GetMessageCount Error:{0}", ex.Message), ex);
                result = -1;
            }

            return result;
        }

 这里说一下:BaseUtil.Logger 是Log4Net的实例对象,另外上面没有针对持续订阅消费消息(ConsumeMsg)作说明,因为这个其实可以不用连接池也不会有问题,因为它是一个持久订阅并持久消费的过程,不会出现频繁创建连接对象的情况。

 最后要说的是,虽说代码贴出来,大家一看就觉得很简单,好像没有什么技术含量,但如果没有完整的思路也还是需要花费一些时间和精力的,代码中核心是如何简单高效的解决并发及连接复用的的问题,该MQHelper有经过压力测试并顺利在我司项目中使用,完美解决了之前的问题,由于这个方案是我在公司通宵实现的,可能有一些方面的不足,大家可以相互交流或完善后入到自己的项目中。