本文實例講述了ActiveMQ在C#中的應用。分享給大家供大家參考，具體如下：
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，網址：http://activemq.apache.org/download.html。我之前下的是5.3.1，5.3.2現在也已經出來了。
2、去ActiveMQ官方網站下載最新版的Apache.NMS，網址：http://activemq.apache.org/nms/download.html，需要下載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; } }}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
msg6-->consumer C
特殊情況是指：ActiveMQ支持過濾機制，即生產者可以設置消息的屬性（Properties），該屬性與消費者端的Selector對應，只有消費者設置的selector與消息的Properties匹配，消息才會發給該消費者。Topic和Queue都支持Selector。
Properties和Selector該如何設置呢？請看如下代碼：
producer：
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：
//生成consumer時通過參數設置SelectorIMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");更多關于C#相關內容感興趣的讀者可查看本站專題：《C#窗體操作技巧匯總》、《C#常見控件用法教程》、《WinForm控件用法總結》、《C#程序設計之線程使用技巧總結》、《C#操作Excel技巧總結》、《C#中XML文件操作技巧匯總》、《C#數據結構與算法教程》、《C#數組操作技巧總結》及《C#面向對象程序設計入門教程》
希望本文所述對大家C#程序設計有所幫助。
新聞熱點
疑難解答
