溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹了C#中關于ActiveMQ的應用示例，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。
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，我之前下的是5.3.1，5.3.2現在也已經出來了。
2、去ActiveMQ官方網站下載最新版的Apache.NMS，需要下載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;
}
}
}
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：
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時通過參數設置Selector
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
//生成consumer時通過參數設置Selector
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");
}感謝你能夠認真閱讀完這篇文章，希望小編分享的“C#中關于ActiveMQ的應用示例”這篇文章對大家有幫助，同時也希望大家多多支持億速云，關注億速云行業資訊頻道，更多相關知識等著你來學習!
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
