最近在開發(fā)過程中由于資源有限凰锡，要對Activemq進(jìn)行高性能處理措近。這里我們只說開發(fā)斯撮。
由下圖可以看到，Activemq是由Connection翎猛、Session、Producer接剩、Consumer切厘、Destination組成。Destination是信息的載體懊缺，通過Producer發(fā)出疫稿，再由Consumer接收。Connection和Session之間是1對多關(guān)系鹃两，Session 和 Producer/Consumer之間也是1對多關(guān)系遗座。程序創(chuàng)建Connection 、Session俊扳、Producer/Consumer都是要消耗服務(wù)器資源的途蒋。正常情況下每個Producer/Consumer都是可以永久使用的。出于網(wǎng)絡(luò)等一些不確定因素造成Producer/Consumer適時失效馋记，這時我們就需要使用到Session池和Producer/Consumer池号坡。在連接恢復(fù)正常后保證數(shù)據(jù)正常提交。這里我提供了10000并發(fā)的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>
    /// 最大并發(fā)
    /// </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>
    /// 默認(rèn)參數(shù)方便用戶進(jìn)行軟件測試
    /// </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>
    /// 創(chuàng)建連接
    /// </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>
    /// 發(fā)送文本內(nèi)容
    /// </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>
    /// 發(fā)送文本內(nèi)容
    /// </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>
    /// 發(fā)送對象
    /// </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>
    /// 發(fā)送對象
    /// </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>
    /// 發(fā)送消息
    /// </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>
    /// 連接異常監(jiān)控
    /// </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>
    /// 生產(chǎn)者
    /// </summary>
    public IMessageProducer Producer { get; set; }
    /// <summary>
    /// 生產(chǎn)者類型
    /// </summary>
    public string Type { get; set; }
    /// <summary>
    /// 生產(chǎn)者名稱
    /// </summary>
    public string Name { get; set; }
    /// <summary>
    /// 當(dāng)前Session產(chǎn)生的會話數(shù)量
    /// </summary>
    public int CurrentSessionCounter { get; set; }
    /// <summary>
    /// Session下Producer計數(shù)器
    /// </summary>
    public bool IsEnable { get; set; } = true;
}