ActiveMQ在C#中的应用

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，网址：http://activemq.apache.org/download.html。我之前下的是5.3.1，5.3.2现在也已经出来了。

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，因此下载最新版即可。

 

 1 　　　　　private void StopMonitorThreads()   
 2         {   
 3             lock(monitor)   
 4             {   
 5                 if(monitorStarted.CompareAndSet(true, false))   
 6                 {   
 7                     AutoResetEvent shutdownEvent = new AutoResetEvent(false);   
 8                     // Attempt to wait for the Timers to shutdown, but don‘t wait   
 9                     // forever, if they don‘t shutdown after two seconds, just quit.   
10                     this.readCheckTimer.Dispose(shutdownEvent);   
11                     shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
12                     this.writeCheckTimer.Dispose(shutdownEvent);   
13                     shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));   
14                                                     //WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext)   
15                     this.asyncTasks.Shutdown();   
16                     this.asyncTasks = null;   
17                     this.asyncWriteTask = null;   
18                     this.asyncErrorTask = null;   
19                 }   
20             }   
21         }  
22 　　　　 private void StopMonitorThreads() 
23         { 
24             lock(monitor) 
25             { 
26                 if(monitorStarted.CompareAndSet(true, false)) 
27                 { 
28                     AutoResetEvent shutdownEvent = new AutoResetEvent(false);
29 
30                     // Attempt to wait for the Timers to shutdown, but don‘t wait 
31                     // forever, if they don‘t shutdown after two seconds, just quit. 
32                     this.readCheckTimer.Dispose(shutdownEvent); 
33                     shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
34                     this.writeCheckTimer.Dispose(shutdownEvent); 
35                     shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 
36                                                     //WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 
37                     this.asyncTasks.Shutdown(); 
38                     this.asyncTasks = null; 
39                     this.asyncWriteTask = null; 
40                     this.asyncErrorTask = null; 
41                 } 
42             } 
43         }

 

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：

 1 using System;   
 2 using System.Collections.Generic;   
 3 using System.Text;   
 4 using Apache.NMS;   
 5 using Apache.NMS.ActiveMQ;   
 6 using System.IO;   
 7 using System.Xml.Serialization;   
 8 using System.Runtime.Serialization.Formatters.Binary;   
 9 namespace Publish   
10 {   
11     class Program   
12     {   
13         static void Main(string[] args)   
14         {   
15             try  
16             {   
17                 //Create the Connection Factory   
18                 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
19                 using (IConnection connection = factory.CreateConnection())   
20                 {   
21                     //Create the Session   
22                     using (ISession session = connection.CreateSession())   
23                     {   
24                         //Create the Producer for the topic/queue   
25                         IMessageProducer prod = session.CreateProducer(   
26                             new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"));   
27                         //Send Messages   
28                         int i = 0;   
29                         while (!Console.KeyAvailable)   
30                         {   
31                             ITextMessage msg = prod.CreateTextMessage();   
32                             msg.Text = i.ToString();   
33                             Console.WriteLine("Sending: " + i.ToString());   
34                             prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);   
35                             System.Threading.Thread.Sleep(5000);   
36                             i++;   
37                         }   
38                     }   
39                 }   
40                 Console.ReadLine();   
41            }   
42             catch (System.Exception e)   
43             {   
44                 Console.WriteLine("{0}",e.Message);   
45                 Console.ReadLine();   
46             }   
47         }   
48     }   
49 } 

consumer:

 

 1 using System;   
 2 using System.Collections.Generic;   
 3 using System.Text;   
 4 using Apache.NMS;   
 5 using Apache.NMS.ActiveMQ;   
 6 using System.IO;   
 7 using System.Xml.Serialization;   
 8 using System.Runtime.Serialization.Formatters.Binary;   
 9 namespace Subscribe   
10 {   
11     class Program   
12     {   
13         static void Main(string[] args)   
14         {   
15             try  
16             {   
17                 //Create the Connection factory   
18                 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");   
19                 //Create the connection   
20                 using (IConnection connection = factory.CreateConnection())   
21                 {   
22                     connection.ClientId = "testing listener";   
23                     connection.Start();   
24                     //Create the Session   
25                     using (ISession session = connection.CreateSession())   
26                     {   
27                         //Create the Consumer   
28                         IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"), "testing listener", null, false);   
29                         consumer.Listener += new MessageListener(consumer_Listener);   
30                         Console.ReadLine();   
31                     }   
32                     connection.Stop();   
33                     connection.Close();   
34                 }   
35             }   
36             catch (System.Exception e)   
37             {   
38                 Console.WriteLine(e.Message);   
39             }   
40         }   
41         static void consumer_Listener(IMessage message)   
42         {   
43             try  
44             {   
45                 ITextMessage msg = (ITextMessage)message;   
46                 Console.WriteLine("Receive: " + msg.Text);   
47            }   
48             catch (System.Exception e)   
49             {   
50                 Console.WriteLine(e.Message);   
51             }   
52         }   
53     }   
54 }  

 

程序实现的功能：生产者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：

 1 public void SetProperties()
 2 {
 3 ITextMessage msg = prod.CreateTextMessage();   
 4                             msg.Text = i.ToString();   
 5                             msg.Properties.SetString("myFilter", "test1");   
 6                             Console.WriteLine("Sending: " + i.ToString());   
 7                             prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);  
 8 ITextMessage msg = prod.CreateTextMessage(); 
 9                             msg.Text = i.ToString(); 
10                             msg.Properties.SetString("myFilter", "test1"); 
11                             Console.WriteLine("Sending: " + i.ToString()); 
12                             prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
13 
14 }

consumer：

1 public void SetSelector()
2 {
3 //生成consumer时通过参数设置Selector   
4 IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter=‘test1‘");  
5 //生成consumer时通过参数设置Selector 
6 IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter=‘test1‘");
7 }

该文章来自博客园：http://www.cnblogs.com/guthing/archive/2010/06/17/1759333.html 

经过简单的整理，感谢guthing

ActiveMQ 其他文章推荐：http://blog.csdn.net/lee353086/article/details/6819123