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++的应用相对麻烦些，稍后写文章介绍。

 

一、去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，所以下载最新版便可。

 

[c-sharp]  view plain copy

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

 

三、运行ActiveMQ，找到ActiveMQ解压后的bin文件夹：.../apache-activemq-5.3.1/bin，执行activemq.bat批处理文件便可启动ActiveMQ服务器，默认端口为61616，这可在配置文件中修改。

 

四、写C#程序实现ActiveMQ的简单应用。新建C#工程（一个Producter项目和一个Consumer项目），WinForm或Console程序都可，这里建的是Console工程，添加对Apache.NMS.dll和Apache.NMS.ActiveMQ.dll的引用，而后便可编写实现代码了，简单的Producer和Consumer实现代码以下：

 

producer：

[c-sharp]  view plain copy

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

 

consumer:

[c-sharp]  view plain copy

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

 

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

[c-sharp]  view plain copy

1. ITextMessage msg = prod.CreateTextMessage();  
2.                             msg.Text = i.ToString();  
3.                             msg.Properties.SetString("myFilter", "test1");  
4.                             Console.WriteLine("Sending: " + i.ToString());  
5.                             prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);  

 

consumer：

[c-sharp]  view plain copy

1. //生成consumer时经过参数设置Selector  
2. IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter='test1'");  

 

------------------------------------------------------------------------------------------------------------------------------------------------------------

介于许多朋友对Demo程序的需求，上传一个以前本身写的WinForm Demo吧，下载地址在这里：http://download.csdn.net/detail/bodybo/4498147

截图以下：

MQ发送程序

 

MQ接收程序