這篇文章主要介紹了C#中關(guān)于ActiveMQ的應(yīng)用示例，具有一定借鑒價(jià)值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

創(chuàng)新互聯(lián)公司服務(wù)項(xiàng)目包括重慶網(wǎng)站建設(shè)、重慶網(wǎng)站制作、重慶網(wǎng)頁(yè)制作以及重慶網(wǎng)絡(luò)營(yíng)銷(xiāo)策劃等。多年來(lái)，我們專(zhuān)注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢(shì)、行業(yè)經(jīng)驗(yàn)、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機(jī)構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，重慶網(wǎng)站推廣取得了明顯的社會(huì)效益與經(jīng)濟(jì)效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到重慶省份的部分城市，未來(lái)相信會(huì)繼續(xù)擴(kuò)大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！

ActiveMQ是個(gè)好東東，不必多說(shuō)。ActiveMQ提供多種語(yǔ)言支持，如Java, C, C++, C#, Ruby, Perl, Python, PHP等。由于我在windows下開(kāi)發(fā)GUI，比較關(guān)心C++和C#，其中C#的ActiveMQ很簡(jiǎn)單，Apache提供NMS（.Net Messaging Service）支持.Net開(kāi)發(fā)，只需如下幾個(gè)步驟即能建立簡(jiǎn)單的實(shí)現(xiàn)。C++的應(yīng)用相對(duì)麻煩些，稍后寫(xiě)文章介紹。

1、去ActiveMQ官方網(wǎng)站下載最新版的ActiveMQ，我之前下的是5.3.1，5.3.2現(xiàn)在也已經(jīng)出來(lái)了。

2、去ActiveMQ官方網(wǎng)站下載最新版的Apache.NMS，需要下載Apache.NMS和Apache.NMS.ActiveMQ兩個(gè)bin包，如果對(duì)源碼感興趣，也可下載src包。這里要提醒一下，如果下載1.2.0版本的NMS.ActiveMQ，Apache.NMS.ActiveMQ.dll在實(shí)際使用中有個(gè)bug，即停止ActiveMQ應(yīng)用時(shí)會(huì)拋WaitOne函數(shù)異常，查看src包中的源碼發(fā)現(xiàn)是由于Apache.NMS.ActiveMQ-1.2.0-src\src\main\csharp\Transport\InactivityMonitor.cs中的如下代碼造成的，修改一下源碼重新編譯即可?？戳艘幌伦钚掳?.3.0已經(jīng)修復(fù)了這個(gè)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、運(yùn)行ActiveMQ，找到ActiveMQ解壓后的bin文件夾：...\apache-activemq-5.3.1\bin，執(zhí)行activemq.bat批處理文件即可啟動(dòng)ActiveMQ服務(wù)器，默認(rèn)端口為61616，這可在配置文件中修改。

4、寫(xiě)C#程序?qū)崿F(xiàn)ActiveMQ的簡(jiǎn)單應(yīng)用。新建C#工程（一個(gè)Producter項(xiàng)目和一個(gè)Consumer項(xiàng)目），WinForm或Console程序均可，這里建的是Console工程，添加對(duì)Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引用，然后即可編寫(xiě)實(shí)現(xiàn)代碼了，簡(jiǎn)單的Producer和Consumer實(shí)現(xiàn)代碼如下：

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

程序?qū)崿F(xiàn)的功能：生產(chǎn)者producer建立名為testing的主題，并每隔5秒向該主題發(fā)送消息，消費(fèi)者consumer訂閱了testing主題，因此只要生產(chǎn)者發(fā)送testing主題的消息到ActiveMQ服務(wù)器，服務(wù)器就將該消息發(fā)送給訂閱了testing主題的消費(fèi)者。

編譯生成producer.exe和consumer.exe，并執(zhí)行兩個(gè)exe，即可看到消息的發(fā)送與接收了。

這個(gè)例子是建的主題（Topic），ActiveMQ還支持另一種方式：Queue，即P2P，兩者有什么區(qū)別呢？區(qū)別在于，Topic是廣播，即如果某個(gè)Topic被多個(gè)消費(fèi)者訂閱，那么只要有消息到達(dá)服務(wù)器，服務(wù)器就將該消息發(fā)給全部的消費(fèi)者；而Queue是點(diǎn)到點(diǎn)，即一個(gè)消息只能發(fā)給一個(gè)消費(fèi)者，如果某個(gè)Queue被多個(gè)消費(fèi)者訂閱，沒(méi)有特殊情況的話消息會(huì)一個(gè)一個(gè)地輪流發(fā)給不同的消費(fèi)者，比如：

msg1-->consumer A

msg2-->consumer B

msg3-->consumer C

msg4-->consumer A

msg5-->consumer B

msg6-->consumer C

特殊情況是指：ActiveMQ支持過(guò)濾機(jī)制，即生產(chǎn)者可以設(shè)置消息的屬性（Properties），該屬性與消費(fèi)者端的Selector對(duì)應(yīng)，只有消費(fèi)者設(shè)置的selector與消息的Properties匹配，消息才會(huì)發(fā)給該消費(fèi)者。Topic和Queue都支持Selector。

Properties和Selector該如何設(shè)置呢？請(qǐng)看如下代碼：

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時(shí)通過(guò)參數(shù)設(shè)置Selector   
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");  
//生成consumer時(shí)通過(guò)參數(shù)設(shè)置Selector 
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
}

感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“C#中關(guān)于ActiveMQ的應(yīng)用示例”這篇文章對(duì)大家有幫助，同時(shí)也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，更多相關(guān)知識(shí)等著你來(lái)學(xué)習(xí)!