最近在开发过程中由于资源有限，要对Activemq进行高性能处理。这里我们只说开发。
由下图可以看到，Activemq是由Connection、Session、Producer、Consumer、Destination组成。Destination是信息的载体，通过Producer发出，再由Consumer接收。Connection和Session之间是1对多关系，Session 和 Producer/Consumer之间也是1对多关系。程序创建Connection 、Session、Producer/Consumer都是要消耗服务器资源的。正常情况下每个Producer/Consumer都是可以永久使用的。出于网络等一些不确定因素造成Producer/Consumer适时失效，这时我们就需要使用到Session池和Producer/Consumer池。在连接恢复正常后保证数据正常提交。这里我提供了10000并发的Producer实例，请大家参考。
image.png
public class MqProducer : BaseMq
{
/// <summary>
/// Session池
/// </summary>
List<ISession> sessions = new List<ISession>();
/// <summary>
/// Producer池
/// </summary>
List<IMessageProducer> Producers = new List<IMessageProducer>();
public List<ProducerPool> ProducerPools { get; set; } = new List<ProducerPool>();
/// <summary>
/// 最大并发
/// </summary>
int SessionCounter = 0;
/// <summary>
/// 持久化
/// </summary>
public bool IsStore { get; set; } = false;
/// <summary>
/// 消息类型 Queue/Topic
/// </summary>
public string messageType { get; set; }
/// <summary>
/// 消息名称
/// </summary>
public string messageName { get; set; }
/// <summary>
/// 默认参数方便用户进行软件测试
/// </summary>
/// <param name="url"></param>
/// <param name="user"></param>
/// <param name="password"></param>
public MqProducer(string url = "activemq:failover:(tcp://127.0.0.1:61616)", string user = "admin", string password = "admin")
{
Url = url;
User = user;
Pwd = password;
factory = new NMSConnectionFactory(Url);
connection = factory.CreateConnection(User, Pwd);
connection.ExceptionListener += ConnectionException;
}
/// <summary>
/// 创建连接
/// </summary>
/// <param name="url"></param>
/// <param name="user"></param>
/// <param name="password"></param>
/// <returns></returns>
public IConnection CreateConnection(string url = "", string user = "", string password = "")
{
Url = string.IsNullOrWhiteSpace(url) ? Url : url;
User = string.IsNullOrWhiteSpace(user) ? User : user;
Pwd = string.IsNullOrWhiteSpace(password) ? Pwd : password;
if (string.IsNullOrWhiteSpace(Url) || string.IsNullOrWhiteSpace(User) || string.IsNullOrWhiteSpace(Pwd)) return null;
factory = new NMSConnectionFactory(Url);
connection = factory.CreateConnection(User, Pwd);
return connection;
}
/// <summary>
///
/// </summary>
/// <param name="type"></param>
/// <param name="name"></param>
/// <returns></returns>
public async Task<IMessageProducer> CreateProducer(string type = "queue", string name = "")
{
messageName = string.IsNullOrWhiteSpace(name) ? messageName : name.ToLower();
messageType = string.IsNullOrWhiteSpace(type) ? messageType : type.ToLower();
var pool = await GetProducerPool(type, name);
return pool.Producer;
}
/// <summary>
/// 发送文本内容
/// </summary>
/// <param name="content"></param>
public async Task SendMessage(string content)
{
ProducerPool producer = await GetProducerPool(messageType, messageName);
await ProducerInUse(producer);
await SendMessage(producer.Producer, producer.Session.CreateTextMessage(content));
await ProducerUnUse(producer);
}
/// <summary>
/// 发送文本内容
/// </summary>
/// <param name="content"></param>
/// <param name="type"></param>
/// <param name="name"></param>
public async Task SendMessage(string content, string type, string name)
{
messageName = string.IsNullOrWhiteSpace(name) ? messageName : name.ToLower();
messageType = string.IsNullOrWhiteSpace(type) ? messageType : type.ToLower();
ProducerPool producer = await GetProducerPool(type, name);
await ProducerInUse(producer);
await SendMessage(producer.Producer, producer.Session.CreateTextMessage(content));
await ProducerUnUse(producer);
}
/// <summary>
/// 发送对象
/// </summary>
/// <param name="content"></param>
public async Task SendMessage(object content)
{
ProducerPool producer = await GetProducerPool(messageType, messageName);
await ProducerInUse(producer);
await SendMessage(producer.Producer, producer.Session.CreateObjectMessage(content));
await ProducerUnUse(producer);
}
/// <summary>
/// 发送对象
/// </summary>
/// <param name="content"></param>
/// <param name="type"></param>
/// <param name="name"></param>
public async Task SendMessage(object content, string type, string name)
{
messageName = string.IsNullOrWhiteSpace(name) ? messageName : name;
messageType = string.IsNullOrWhiteSpace(type) ? messageType : type;
ProducerPool producer = await GetProducerPool(type, name);
await ProducerInUse(producer);
await SendMessage(producer.Producer, producer.Session.CreateObjectMessage(content));
await ProducerUnUse(producer);
}
private async Task ProducerInUse(ProducerPool pool)
{
await Task.Run(() =>
{
int index = ProducerPools.IndexOf(pool);
ProducerPools[index].IsEnable = false;
});
}
private async Task ProducerUnUse(ProducerPool pool)
{
await Task.Run(() =>
{
int index = ProducerPools.IndexOf(pool);
ProducerPools[index].IsEnable = true;
});
}
public async Task<ProducerPool> GetProducerPool(string type, string name)
{
if (string.IsNullOrWhiteSpace(type) || string.IsNullOrWhiteSpace(name)) return null;
var canUseProducers = ProducerPools.Where(m => m.IsEnable == true && m.Type == type.ToLower() && m.Name == name.ToLower());
var canUseSessions = ProducerPools.Where(m => m.CurrentSessionCounter < 100);
if (canUseProducers.Count() == 0 && canUseSessions.Count() > 0)
{
ProducerPool producerPool = ProducerPools.Where(m => m.CurrentSessionCounter < 100).FirstOrDefault();
producerPool = await CreateProducerPool(producerPool.Session, type.ToLower(), name.ToLower());
return producerPool;
}
else if (canUseSessions.Count() == 0 && SessionCounter < 100)
{
ProducerPool producerPool = await CreateSessionPool();
producerPool = await CreateProducerPool(producerPool.Session, type.ToLower(), name.ToLower());
return producerPool;
}
else
{
while (canUseProducers.Count() == 0)
{
canUseProducers = ProducerPools.Where(m => m.IsEnable == true && m.Type == type.ToLower() && m.Name == name.ToLower());
}
return canUseProducers.FirstOrDefault();
}
}
private async Task<ProducerPool> CreateProducerPool(ISession session, string type, string name)
{
IMessageProducer producer = await CreateProducer(session, type, name);
ProducerPool producerPool = new ProducerPool() { Session = session, Producer = producer, Type = type.ToLower(), Name = name.ToLower(), CurrentSessionCounter = 0 };
ProducerPools.Insert(0, producerPool);
return producerPool;
}
private async Task<ProducerPool> CreateSessionPool()
{
ISession session = await CreateSession();
ProducerPool producerPool = new ProducerPool() { Session = session, CurrentSessionCounter = 0 };
return producerPool;
}
private IDestination CreateDestination(ISession session, string type, string name)
{
IDestination destination = null;
if (session == null || string.IsNullOrWhiteSpace(type) || string.IsNullOrWhiteSpace(name)) return destination;
if (type == "topic") { destination = session.GetTopic(name); }
else { destination = session.GetQueue(name); }
return destination;
}
private async Task<IMessageProducer> CreateProducer(ISession session, string type, string name)
{
return await Task.Run(() =>
{
IDestination dest = CreateDestination(session, type, name);
return session.CreateProducer(dest);
});
}
private async Task<ISession> CreateSession()
{
return await Task.Run(() =>
{
if (connection == null) CreateConnection();
ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
SessionCounter++;
return session;
});
}
/// <summary>
/// 发送消息
/// </summary>
/// <param name="producer"></param>
/// <param name="message"></param>
/// <returns></returns>
public async Task SendMessage(IMessageProducer producer, IMessage message)
{
await Task.Run(() =>
{
producer.Send(message);
});
}
/// <summary>
/// 连接异常监控
/// </summary>
/// <param name="ex"></param>
private void ConnectionException(Exception ex)
{
MessageBox.Show(ex.Message + "\r\n" + ex.StackTrace);
}
}
public class ProducerPool
{
/// <summary>
/// Activemq Session
/// </summary>
public ISession Session { get; set; }
/// <summary>
/// 生产者
/// </summary>
public IMessageProducer Producer { get; set; }
/// <summary>
/// 生产者类型
/// </summary>
public string Type { get; set; }
/// <summary>
/// 生产者名称
/// </summary>
public string Name { get; set; }
/// <summary>
/// 当前Session产生的会话数量
/// </summary>
public int CurrentSessionCounter { get; set; }
/// <summary>
/// Session下Producer计数器
/// </summary>
public bool IsEnable { get; set; } = true;
}
网友评论