activemq讀取剩餘消息隊列中消息的數量

先上原文連接：html

http://blog.csdn.net/bodybo/article/details/5647968 ActiveMQ在C#中的應用apache

ActiveMQ是個好東東，沒必要多說。ActiveMQ提供多種語言支持，如Java, C, C++, C#, Ruby, Perl, Python, PHP等。因爲我在windows下開發GUI，比較關心C++和C#，其中C#的ActiveMQ很簡單，Apache提供NMS（.Net Messaging Service）支持.Net開發，只需以下幾個步驟即能創建簡單的實現。C++的應用相對麻煩些，稍後寫文章介紹。windows

一、去ActiveMQ官方網站下載最新版的ActiveMQ，網址：http://activemq.apache.org/download.html。我以前下的是5.3.1，5.3.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，所以下載最新版便可。session

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

三、運行ActiveMQ，找到ActiveMQ解壓後的bin文件夾：.../apache-activemq-5.3.1/bin，執行activemq.bat批處理文件便可啓動ActiveMQ服務器，默認端口爲61616，這可在配置文件中修改。async

四、寫C#程序實現ActiveMQ的簡單應用。新建C#工程（一個Producter項目和一個Consumer項目），WinForm或Console程序都可，這裏建的是Console工程，添加對Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引用，而後便可編寫實現代碼了，簡單的Producer和Consumer實現代碼以下：tcp

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，便可看到消息的發送與接收了。ui

這個例子是建的主題（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該如何設置呢？請看以下代碼：

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

producer：

//生成consumer時經過參數設置Selector  
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");

介於許多朋友對Demo程序的需求，上傳一個以前本身寫的WinForm Demo吧，下載地址在這裏：http://download.csdn.net/detail/bodybo/4498147