C#中关于ActiveMQ的应用详解

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的应用详解的详细内容，更多请关注其它相关文章！

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！