基于RabbitMQ.Client组件实现RabbitMQ可复用的 ConnectionPool（连接池）
程序员文章站
2022-06-18 09:11:43
一、本文产生原由： 之前文章《总结消息队列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有经过压力测试并顺利在我司项目中使用，完美解决了之前的问题，由于这个方案是我在公司通宵实现的，可能有一些方面的不足，大家可以相互交流或完善后入到自己的项目中。