本文實例講述了C#實現同Active MQ通訊的方法。分享給大家供大家參考，具體如下：
內容概要：
主要以源碼的形式介紹如何用C#實現同Active MQ 的通訊。本文假設你已經正確安裝JDK1.6.x，了解Active MQ并有一定的編程基礎。
正文：
JMS 程序的最終目的是生產和消費的消息能被其他程序使用，JMS 的 Message 是一個既簡單又不乏靈活性的基本格式，允許創建不同平臺上符合非JMS 程序格式的消息。
Message 由消息頭，屬性和消息體三部份組成。
Active MQ支持過濾機制，即生產者可以設置消息的屬性（Properties），該屬性與消費者端的Selector對應，只有消費者設置的selector與消息的Properties匹配，消息才會發給該消費者。Topic和Queue都支持Selector。
示例代碼：
using System;using System.Collections.Generic;using System.Linq;using System.Text;using System.Windows;using System.Windows.Controls;using System.Windows.Data;using System.Windows.Documents;using System.Windows.Input;using System.Windows.Media;using System.Windows.Media.Imaging;using System.Windows.Navigation;using System.Windows.Shapes;using Apache.NMS;using System.Diagnostics;using Apache.NMS.Util;using System.Windows.Threading;/* * 功能描述：C#使用ActiveMQ示例 * 修改次數：2 * 最后更新： by Kagula,2012-07-31 * * 前提條件： * [1]apache-activemq-5.4.2 * [2]Apache.NMS.ActiveMQ-1.5.6-bin * [3]WinXP SP3 * [4]VS2008 SP1 * [5]WPF工程 With .NET Framework 3.5 * * 啟動 * * 不帶安全控制方式啟動 * [你的解壓路徑]/apache-activemq-5.4.2/bin/activemq.bat * * 安全方式啟動 * 添加環境變量： ACTIVEMQ_ENCRYPTION_PASSWORD=activemq * [你的解壓路徑]/apache-activemq-5.4.2/bin>activemq xbean:file:../conf/activemq-security.xml * * Active MQ 管理地址 * http://127.0.0.1:8161/admin/ * 添加訪問"http://127.0.0.1:8161/admin/"的限制 * * 第一步：添加訪問限制 * 修改D:/apache/apache-activemq-5.4.2/conf/jetty.xml文件 * 下面這行編碼，原 * <property name="authenticate" value="true" /> * 修改為 * <property name="authenticate" value="false" /> * * 第二步：修改登錄用戶名密碼，缺省分別為admin，admin * D:/apache/apache-activemq-5.4.2/conf/jetty-realm.properties * * 用戶管理(前提:以安全方式啟動ActiveMQ) * * 在[你的解壓路徑]/apache-activemq-5.4.2/conf/credentials.properties文件中修改默認的用戶名密碼 * 在[你的解壓路徑]/apache-activemq-5.4.2/conf/activemq-security.xml文件中可以添加新的用戶名 * e.g. 添加oa用戶，密碼同用戶名。 * <authenticationUser username="oa" password="oa" groups="users,admins"/> * * 在[你的解壓路徑]/apache-activemq-5.4.2/conf/activemq-security.xml文件中你還可以設置指定的Topic或Queue * 只能被哪些用戶組read 或 write。 * * * 配置C# with WPF項目 * 項目的[Application]->[TargetFramework]屬性設置為[.NETFramework 3.5](這是VS2008WPF工程的默認設置) * 添加[你的解壓路徑]/Apache.NMS.ActiveMQ-1.5.6-bin/lib/Apache.NMS/net-3.5/Apache.NMS.dll的引用 * Apache.NMS.dll相當于接口 * * 如果是以Debug方式調試 * 把[你的解壓路徑]/Apache.NMS.ActiveMQ-1.5.6-bin/build/net-3.5/debug/目錄下的 * Apache.NMS.ActiveMQ.dll文件復制到你項目的Debug目錄下 * Apache.NMS.ActiveMQ.dll相當于實現 * * 如果是以Release方式調試 * 參考上文，去取Apache.NMS,Release目錄下相應的DLL文件，并復制到你項目的Release目錄下。 * * * 參考資料 * [1]《C#調用ActiveMQ官方示例》 http://activemq.apache.org/nms/examples.html * [2]《ActiveMQ NMS下載地址》http://activemq.apache.org/nms/activemq-downloads.html * [3]《Active MQ在C#中的應用示例》//www.49028c.com/article/87956.htm * [4]《NMS API Reference》http://activemq.apache.org/nms/nms-api.html */namespace testActiveMQSubscriber{ /// <summary> /// Interaction logic for Window1.xaml /// </summary> public partial class Window1 : Window { private static IConnectionFactory connFac; private static IConnection connection; private static ISession session; private static IDestination destination; private static IMessageProducer producer; private static IMessageConsumer consumer; protected static ITextMessage message = null; public Window1() { InitializeComponent(); initAMQ("MyFirstTopic"); } private void initAMQ(String strTopicName) { try { connFac = new NMSConnectionFactory(new Uri("activemq:failover:(tcp://localhost:61616)")); //新建連接 //connection = connFac.CreateConnection("oa","oa");//設置連接要用的用戶名、密碼 //如果你要持久“訂閱”，則需要設置ClientId，這樣程序運行當中被停止，恢復運行時，能拿到沒接收到的消息！ connection.ClientId = "testing listener"; connection = connFac.CreateConnection();//如果你是缺省方式啟動Active MQ服務，則不需填用戶名、密碼 //創建Session session = connection.CreateSession(); //發布/訂閱模式，適合一對多的情況 destination = SessionUtil.GetDestination(session, "topic://" + strTopicName); //新建生產者對象 producer = session.CreateProducer(destination); producer.DeliveryMode = MsgDeliveryMode.NonPersistent;//ActiveMQ服務器停止工作后，消息不再保留 //新建消費者對象:普通“訂閱”模式 //consumer = session.CreateConsumer(destination);//不需要持久“訂閱” //新建消費者對象:持久"訂閱"模式： // 持久“訂閱”后，如果你的程序被停止工作后，恢復運行， //從第一次持久訂閱開始，沒收到的消息還可以繼續收 consumer = session.CreateDurableConsumer( session.GetTopic(strTopicName) , connection.ClientId, null, false); //設置消息接收事件 consumer.Listener += new MessageListener(OnMessage); //啟動來自Active MQ的消息偵聽 connection.Start(); } catch (Exception e) { //初始化ActiveMQ連接失敗，往VS2008的Output窗口寫入出錯信息! Debug.WriteLine(e.Message); } } private void SendMsg2Topic_Click(object sender, RoutedEventArgs e) { //發送消息 ITextMessage request = session.CreateTextMessage(DateTime.Now.ToLocalTime()+" "+tbMsg.Text); producer.Send(request); } protected void OnMessage(IMessage receivedMsg) { //接收消息 message = receivedMsg as ITextMessage; //UI線程，顯示收到的消息 Dispatcher.Invoke(DispatcherPriority.Normal, new Action(() => { DateTime dt = new DateTime(); ListBoxItem lbi = new ListBoxItem(); lbi.Content = DateTime.Now.ToLocalTime() + " " + message.Text; lbR.Items.Add(lbi); })); } }}隊列通訊方式，消費者例子
using System;using System.Collections.Generic;using System.Linq;using System.Text;using Apache.NMS;using System.Diagnostics;using log4net;using Apache.NMS.Util;using System.Collections;namespace Cat8637AutoCallServer{ public class SMTask { public String Callee { get; set; } public String CheckNumber { get; set; } public int Deadline { get; set; } public override String ToString() { return String.Format("Callee={0},CheckNumber={1},Deadline={2}", Callee,CheckNumber,Deadline); } } /* * 負責接收任務，并把任務放在任務等待隊列中。 */ public class MQClient { private static readonly ILog logger = LogManager.GetLogger(typeof(MQClient)); private static IConnection connection = null; private static ISession session = null; Queue _voiceSMTasks = new Queue(); public MQClient() { try { IConnectionFactory factory = new NMSConnectionFactory(new Uri("activemq:failover:(tcp://localhost:61616)")); //新建連接 //connection = connFac.CreateConnection("oa","oa");//設置連接要用的用戶名、密碼 connection = factory.CreateConnection(); session = connection.CreateSession(); IMessageConsumer consumer = session.CreateConsumer(session.GetQueue("TaskIssue_VoiceSM")); consumer.Listener += new MessageListener(OnMessage); connection.Start(); } catch (Exception ex) { Debug.WriteLine(ex.Message); } } protected void OnMessage(IMessage receivedMsg) { IMessage message = receivedMsg as ITextMessage; SMTask smTask = new SMTask(); smTask.Callee = message.Properties["Callee"] as String; smTask.CheckNumber = message.Properties["Message"] as String; smTask.Deadline = Convert.ToInt32(message.Properties["deadline"] as String); logger.Info("Received: "+smTask.ToString()); lock (_voiceSMTasks) { _voiceSMTasks.Enqueue(smTask); } } public SMTask GetVoiceSMTask() { SMTask result = null; lock (_voiceSMTasks) { if (_voiceSMTasks.Count > 0) { result = _voiceSMTasks.Dequeue() as SMTask; } } return result; } }}隊列通訊方式，生產者例子
private void Send_Click(object sender, RoutedEventArgs e){ try { IDestination destination = SessionUtil.GetDestination(session, "queue://TaskIssue_VoiceSM"); //新建生產者對象 IMessageProducer producer = session.CreateProducer(destination); producer.DeliveryMode = MsgDeliveryMode.NonPersistent;//ActiveMQ服務器停止工作后，消息不再保留 ITextMessage request = session.CreateTextMessage(); request.NMSCorrelationID = "TestVoiceSM";//這里我填了應用程序的名稱。 request.Properties["Callee"] = tbCallee.Text; request.Properties["Message"] = tbCheckNumber.Text; request.Properties["deadline"] = tbValidDuration.Text; producer.Send(request); } catch (Exception ex) { //初始化ActiveMQ連接失敗，往VS2008的Output窗口寫入出錯信息! Debug.WriteLine(ex.Message); }}private void Window_Closed(object sender, EventArgs e){ try { if (session == null) return; //if (connection == null) // return; session.Close(); //connection.Close(); } catch (Exception ex) { Debug.WriteLine(ex.Message); }}更多關于C#相關內容感興趣的讀者可查看本站專題：《C#窗體操作技巧匯總》、《C#常見控件用法教程》、《WinForm控件用法總結》、《C#程序設計之線程使用技巧總結》、《C#操作Excel技巧總結》、《C#中XML文件操作技巧匯總》、《C#數據結構與算法教程》、《C#數組操作技巧總結》及《C#面向對象程序設計入門教程》
希望本文所述對大家C#程序設計有所幫助。
新聞熱點
疑難解答
