當(dāng)前位置：首頁(yè) > 人文社科 > 生活经验 >内容正文

生活经验

ActiveMQ在C#中的应用

發(fā)布時(shí)間：2023/11/27 生活经验 42 豆豆

生活随笔收集整理的這篇文章主要介紹了 ActiveMQ在C#中的应用小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

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，網(wǎng)址：http://activemq.apache.org/download.html。我之前下的是5.3.1，5.3.2現(xiàn)在也已經(jīng)出來(lái)了。

2、去ActiveMQ官方網(wǎng)站下載最新版的Apache.NMS，網(wǎng)址：http://activemq.apache.org/nms/download.html，需要下載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中的如下代碼造成的，修改一下源碼重新編譯即可。看了一下最新版1.3.0已經(jīng)修復(fù)了這個(gè)bug，因此下載最新版即可。

view plaincopy to clipboardprint?
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：

view plaincopy to clipboardprint?
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();??
??????????? }??
??????? }??
??? }??
}?
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:

view plaincopy to clipboardprint?
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);??
??????????? }??
??????? }??
??? }??
}?
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：

view plaincopy to clipboardprint?
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：

view plaincopy to clipboardprint?
//生成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'");

本文來(lái)自CSDN博客，出處：http://blog.csdn.net/bodybo/archive/2010/06/04/5647968.aspx

轉(zhuǎn)載于:https://www.cnblogs.com/guthing/archive/2010/06/17/1759333.html

總結(jié)

以上是生活随笔為你收集整理的ActiveMQ在C#中的应用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

ActiveMQ

上一篇：求一个好听的qq空间名字！
下一篇： printf(%d, -10u); 这个