ActiveMQ在C#中的應用示例分析

更新時間：2016年07月05日 11:44:26 作者：wolfman

這篇文章主要介紹了ActiveMQ在C#中的應用,簡單介紹了ActiveMQ的功能、下載及在C#中的具體使用方法,需要的朋友可以參考下

本文實例講述了ActiveMQ在C#中的應用。分享給大家供大家參考，具體如下：

ActiveMQ是個好東東，不必多說。ActiveMQ提供多種語言支持，如Java, C, C++, C#, Ruby, Perl, Python, PHP等。由于我在windows下開發(fā)GUI，比較關心C++和C#，其中C#的ActiveMQ很簡單，Apache提供NMS（.Net Messaging Service）支持.Net開發(fā)，只需如下幾個步驟即能建立簡單的實現(xiàn)。C++的應用相對麻煩些，后面會有文章介紹。

1、去ActiveMQ官方網站下載最新版的ActiveMQ，網址：http://activemq.apache.org/download.html。我之前下的是5.3.1，5.3.2現(xiàn)在也已經出來了。

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函數(shù)異常，查看src包中的源碼發(fā)現(xiàn)是由于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，執(zhí)行activemq.bat批處理文件即可啟動ActiveMQ服務器，默認端口為61616，這可在配置文件中修改。

4、寫C#程序實現(xiàn)ActiveMQ的簡單應用。新建C#工程（一個Producter項目和一個Consumer項目），WinForm或Console程序均可，這里建的是Console工程，添加對Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引用，然后即可編寫實現(xiàn)代碼了，簡單的Producer和Consumer實現(xiàn)代碼如下：

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);
      }
    }
  }
}

程序實現(xiàn)的功能：生產者producer建立名為testing的主題，并每隔5秒向該主題發(fā)送消息，消費者consumer訂閱了testing主題，因此只要生產者發(fā)送testing主題的消息到ActiveMQ服務器，服務器就將該消息發(fā)送給訂閱了testing主題的消費者。

編譯生成producer.exe和consumer.exe，并執(zhí)行兩個exe，即可看到消息的發(fā)送與接收了。

這個例子是建的主題（Topic），ActiveMQ還支持另一種方式：Queue，即P2P，兩者有什么區(qū)別呢？區(qū)別在于，Topic是廣播，即如果某個Topic被多個消費者訂閱，那么只要有消息到達服務器，服務器就將該消息發(fā)給全部的消費者；而Queue是點到點，即一個消息只能發(fā)給一個消費者，如果某個Queue被多個消費者訂閱，沒有特殊情況的話消息會一個一個地輪流發(fā)給不同的消費者，比如：

msg1-->consumer A

msg2-->consumer B

msg3-->consumer C

msg4-->consumer A

msg5-->consumer B

msg6-->consumer C

特殊情況是指：ActiveMQ支持過濾機制，即生產者可以設置消息的屬性（Properties），該屬性與消費者端的Selector對應，只有消費者設置的selector與消息的Properties匹配，消息才會發(fā)給該消費者。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時通過參數(shù)設置Selector
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");

更多關于C#相關內容感興趣的讀者可查看本站專題：《C#窗體操作技巧匯總》、《C#常見控件用法教程》、《WinForm控件用法總結》、《C#程序設計之線程使用技巧總結》、《C#操作Excel技巧總結》、《C#中XML文件操作技巧匯總》、《C#數(shù)據結構與算法教程》、《C#數(shù)組操作技巧總結》及《C#面向對象程序設計入門教程》

希望本文所述對大家C#程序設計有所幫助。

您可能感興趣的文章: