ActiveMQ在C#中的運用示例剖析。本站提示廣大學習愛好者：（ActiveMQ在C#中的運用示例剖析）文章只能為提供參考，不一定能成為您想要的結果。以下是ActiveMQ在C#中的運用示例剖析正文
本文實例講述了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時經由過程參數設置Selector
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
願望本文所述對年夜家C#法式設計有所贊助。
