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

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

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時(shí)通過參數(shù)設(shè)置Selector
IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");