C#中關於ActiveMQ的應用詳解

ActiveMQ是個好東東，不必多說。 ActiveMQ提供多種語言支持，如Java, C, C++, C#, Ruby, Perl, Python, PHP等。由於我在windows下開發GUI，比較關心C++和C#，其中C#的ActiveMQ很簡單，Apache提供NMS（.Net Messaging Service）支援.Net開發，只需如下幾個步驟即能建立簡單的實作。 C++的應用相對麻煩些，稍後再寫文章介紹。

1、去ActiveMQ官方網站下載最新版的ActiveMQ，我之前下的是5.3.1，5.3.2現在也已經出來了。

2、去ActiveMQ官方網站下載最新版的Apache.NMS，需要下載Apache.NMS和Apache.NMS.ActiveMQ兩個bin包，如果對原始碼有興趣，也可下載src包。這裡要提醒一下，如果下載1.2.0版本的NMS.ActiveMQ，Apache.NMS.ActiveMQ.dll在實際使用中有個bug，即停止ActiveMQ應用時會拋WaitOne函數異常，查看src包中的源碼發現是由於Apache.NMS.ActiveMQ-1.2.0-src\src\main\csharp\Transport\InactivityMonitor.cs中的如下程式碼造成的，修改一下原始碼重新編譯即可。看了一下最新版1.3.0已經修復了這個bug，所以下載最新版即可。

private void StopMonitorThreads()   
        {   
            lock(monitor)   
            {   
                if(monitorStarted.CompareAndSet(true, false))   
                {   
                    AutoResetEvent shutdownEvent = new AutoResetEvent(false);   
                    // Attempt to wait for the Timers to shutdown, but don't wait   
                    // forever, if they don't shutdown after two seconds, just quit.   
                    this.readCheckTimer.Dispose(shutdownEvent);   
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
                    this.writeCheckTimer.Dispose(shutdownEvent);   
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
                                                    //WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext)   
                    this.asyncTasks.Shutdown();   
                    this.asyncTasks = null;   
                    this.asyncWriteTask = null;   
                    this.asyncErrorTask = null;   
                }   
            }   
        }  
　　　　 private void StopMonitorThreads() 
        { 
            lock(monitor) 
            { 
                if(monitorStarted.CompareAndSet(true, false)) 
                { 
                    AutoResetEvent shutdownEvent = new AutoResetEvent(false);

                    // Attempt to wait for the Timers to shutdown, but don't wait 
                    // forever, if they don't shutdown after two seconds, just quit. 
                    this.readCheckTimer.Dispose(shutdownEvent); 
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
                    this.writeCheckTimer.Dispose(shutdownEvent); 
                    shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
                                                    //WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 
                    this.asyncTasks.Shutdown(); 
                    this.asyncTasks = null; 
                    this.asyncWriteTask = null; 
                    this.asyncErrorTask = null; 
                } 
            } 
        }

3、運行ActiveMQ，找到ActiveMQ解壓縮後的bin資料夾：...\apache-activemq-5.3.1\bin，執行activemq.bat批次檔即可啟動ActiveMQ伺服器，預設端口為61616，這可在設定檔中修改。

4、寫C#程式實作ActiveMQ的簡單應用。新建C#工程（一個Producter專案和一個Consumer專案），WinForm或Console程式皆可，這裡建的是Console工程，新增對Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引用，然後即可編寫實作程式碼了，簡單的Producer和Consumer實作程式碼如下：

producer：

using System;   
using System.Collections.Generic;   
using System.Text;   
using Apache.NMS;   
using Apache.NMS.ActiveMQ;   
using System.IO;   
using System.Xml.Serialization;   
using System.Runtime.Serialization.Formatters.Binary;   
namespace Publish   
{   
    class Program   
    {   
        static void Main(string[] args)   
        {   
            try  
            {   
                //Create the Connection Factory   
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
                using (IConnection connection = factory.CreateConnection())   
                {   
                    //Create the Session   
                    using (ISession session = connection.CreateSession())   
                    {   
                        //Create the Producer for the topic/queue   
                        IMessageProducer prod = session.CreateProducer(   
                            new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"));   
                        //Send Messages   
                        int i = 0;   
                        while (!Console.KeyAvailable)   
                        {   
                            ITextMessage msg = prod.CreateTextMessage();   
                            msg.Text = i.ToString();   
                            Console.WriteLine("Sending: " + i.ToString());   
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);   
                            System.Threading.Thread.Sleep(5000);   
                            i++;   
                        }   
                    }   
                }   
                Console.ReadLine();   
           }   
            catch (System.Exception e)   
            {   
                Console.WriteLine("{0}",e.Message);   
                Console.ReadLine();   
            }   
        }   
    }   
}

consumer:

using System;   
using System.Collections.Generic;   
using System.Text;   
using Apache.NMS;   
using Apache.NMS.ActiveMQ;   
using System.IO;   
using System.Xml.Serialization;   
using System.Runtime.Serialization.Formatters.Binary;   
namespace Subscribe   
{   
    class Program   
    {   
        static void Main(string[] args)   
        {   
            try  
            {   
                //Create the Connection factory   
                IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
                //Create the connection   
                using (IConnection connection = factory.CreateConnection())   
                {   
                    connection.ClientId = "testing listener";   
                    connection.Start();   
                    //Create the Session   
                    using (ISession session = connection.CreateSession())   
                    {   
                        //Create the Consumer   
                        IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"), "testing listener", null, false);   
                        consumer.Listener += new MessageListener(consumer_Listener);   
                        Console.ReadLine();   
                    }   
                    connection.Stop();   
                    connection.Close();   
                }   
            }   
            catch (System.Exception e)   
            {   
                Console.WriteLine(e.Message);   
            }   
        }   
        static void consumer_Listener(IMessage message)   
        {   
            try  
            {   
                ITextMessage msg = (ITextMessage)message;   
                Console.WriteLine("Receive: " + msg.Text);   
           }   
            catch (System.Exception e)   
            {   
                Console.WriteLine(e.Message);   
            }   
        }   
    }   
}

程式實作的功能：生產者producer建立一個名為testing的主題，並每隔5秒向該主題發送訊息，消費者consumer訂閱了testing主題，因此只要生產者發送testing主題的訊息到ActiveMQ伺服器，伺服器就將該訊息發送給訂閱了testing主題的消費者。

編譯產生producer.exe和consumer.exe，並執行兩個exe，即可​​看到訊息的傳送與接收了。

這個範例是建造的主題（Topic），ActiveMQ還支援另一種方式：Queue，也就是P2P，兩者有什麼差別呢？差別在於，Topic是廣播，也就是如果某個Topic被多個消費者訂閱，那麼只要有訊息到達伺服器，伺服器就將該訊息發給全部的消費者；而Queue是點到點，也就是一個訊息只能發給一個消費者，如果某個Queue被多個消費者訂閱，沒有特殊情況的話消息會一個一個地輪流發給不同的消費者，比如：

msg1-->consumer A

msg2-->consumer B

msg3-->consumer C

msg4-->consumer A

msg5-->consumer B

##msg4-->consumer A

##msg5-->consumer B

msg6-->consumer C

特殊情況是指：ActiveMQ支援過濾機制，即生產者可以設定訊息的屬性（Properties），該屬性與消費者端的Selector對應，只有消費者設定的selector與訊息的Properties匹配，訊息才會發給該消費者。 Topic和Queue都支援Selector。

Properties和Selector該如何設定？請看如下程式碼：

producer：

public void SetProperties()
{
ITextMessage msg = prod.CreateTextMessage();   
                            msg.Text = i.ToString();   
                            msg.Properties.SetString("myFilter", "test1");   
                            Console.WriteLine("Sending: " + i.ToString());   
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);  
ITextMessage msg = prod.CreateTextMessage(); 
                            msg.Text = i.ToString(); 
                            msg.Properties.SetString("myFilter", "test1"); 
                            Console.WriteLine("Sending: " + i.ToString()); 
                            prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);

}

###consumer：###

public void SetSelector()
{
//生成consumer时通过参数设置Selector   
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");  
//生成consumer时通过参数设置Selector 
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
}

以上是C#中關於ActiveMQ的應用詳解的詳細內容。更多資訊請關注PHP中文網其他相關文章！