ActiveMQ初體驗

首先介紹下MQ，MQ英文名MessageQueue，中文名也就是大家用的消息隊列，干嘛用的呢，說白了就是一個消息的接受和轉發的容器，可用於消息推送。

下面介紹主題，就是今天為大家介紹的ActiveMQ：

他是Apache出品的一個開源的消息隊列軟件，運行在JVM下，支持多種語言，如JAVA，C++，C#。

現在先為大家介紹下如何配置ActiveMQ的服務器端：

1、當然是下載軟件了

去官方網站下載：http://activemq.apache.org/ 我下載的是apache-activemq-5.8.0-bin 5.8版本，當然開源的也是支持下載source的，需要自己編譯下，這里不做過多介紹

2、解壓后，進入\apache-activemq-5.8.0\bin\win32啟動activemq.bat。系統會自動執行啟動過程，當然一般安裝失敗的情況是沒有裝JVM環境，啟動成功應該是這樣

3、打開瀏覽器輸入http://localhost:8161/admin/默認配置是這個，當然你也可以更改這個配置

 

 

4、至此，服務端啟動完畢

—————————————————————————————————完美分割———————————————————————————————————

現在就是今天的主題了，怎樣在C#中使用ActiveMQ提供的API實現消息的訂閱和發布

1、首先需要下載ActiveMQ提供的API函數，這個也從官網下Apache.NMS-1.6.0-bin和Apache.NMS.ActiveMQ-1.6.0-bin 這兩個dll都在下載后文件夾的bin目錄下；

2、將這兩個文件在項目中引用；

3、在ActiveMQ中，有兩個概念，一個是生產者（Producer），另一個是消費者（Consumer），生產者就是我們常說的發布者，而消費者，就是訂閱者，這樣解釋可能更好理解一下吧，如果知道發布訂閱模式的話，不知道的話，字面意思也很好理解，發布者就是發布消息的，而訂閱者通過訂閱事件，將消息接收到；

4、直接上代碼了，Winform下的代碼，如果不想處理界面線程回調問題，可以使用Console程序

5、Produce

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using Apache.NMS;
using Apache.NMS.ActiveMQ;

namespace MqProducer
{
    public partial class ProducerDemo : Form
    {
        //聲明連接對象工廠
        private IConnectionFactory factory;

        public ProducerDemo()
        {
            InitializeComponent();
            InitProducer();
        }

        public void InitProducer()
        {
            try
            {
                //初始化工廠，這里默認的URL是不需要修改的
                factory = new ConnectionFactory("tcp://localhost:61616");

            }
            catch
            {
                lbMessage.Text = "初始化失敗!!";
            }
        }

        private void btnConfirm_Click(object sender, EventArgs e)
        {
            //通過工廠建立連接
            using (IConnection connection = factory.CreateConnection())
            {
                //通過連接創建Session會話
                using (ISession session = connection.CreateSession())
                {
                    //通過會話創建生產者，方法里面new出來的是MQ中的Queue
                    IMessageProducer prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("firstQueue"));
                    //創建一個發送的消息對象
                    ITextMessage message = prod.CreateTextMessage();
                    //給這個對象賦實際的消息
                    message.Text = txtMessage.Text;
                    //設置消息對象的屬性，這個很重要哦，是Queue的過濾條件，也是P2P消息的唯一指定屬性
                    message.Properties.SetString("filter","demo");
                    //生產者把消息發送出去，幾個枚舉參數MsgDeliveryMode是否長鏈，MsgPriority消息優先級別，發送最小單位，當然還有其他重載
                    prod.Send(message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
                    lbMessage.Text = "發送成功!!";
                    txtMessage.Text = "";
                    txtMessage.Focus();
                }
            }

        }
    }
}

6、consumer

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using Apache.NMS;
using Apache.NMS.ActiveMQ;

namespace MqConsumer
{
    public partial class ConsumerDemo : Form
    {
        public ConsumerDemo()
        {
            InitializeComponent();
            InitConsumer();
        }

        public void InitConsumer()
        {
            //創建連接工廠
            IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616");
            //通過工廠構建連接
            IConnection connection = factory.CreateConnection();
            //這個是連接的客戶端名稱標識
            connection.ClientId = "firstQueueListener";
            //啟動連接，監聽的話要主動啟動連接
            connection.Start();
            //通過連接創建一個會話
            ISession session = connection.CreateSession();
            //通過會話創建一個消費者，這里就是Queue這種會話類型的監聽參數設置
            IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("firstQueue"), "filter='demo'");
            //注冊監聽事件
            consumer.Listener += new MessageListener(consumer_Listener);
            //connection.Stop();
            //connection.Close();  

        }

        void consumer_Listener(IMessage message)
        {
            ITextMessage msg = (ITextMessage)message;
            //異步調用下，否則無法回歸主線程
            tbReceiveMessage.Invoke(new DelegateRevMessage(RevMessage),msg);

        }

        public delegate void DelegateRevMessage(ITextMessage message);

        public void RevMessage(ITextMessage message)
        {
            tbReceiveMessage.Text += string.Format(@"接收到:{0}{1}", message.Text, Environment.NewLine);
        }
    }
}

7、啟動界面這就完事了

　　

今天咱說的就是一個皮毛，我也是下午接到任務要做MQ方面的開發，才臨時抱的佛教，希望拋磚引玉，大家繼續鑽研，當然沒事可以把API的代碼下來自己看看以上是怎么實現的，開源的好處就不多說了，自己做功課去了