Activemq 高性能开发总结

最近在开发过程中由于资源有限，要对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实例，请大家参考。

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;
}