ActiveMQ在C#中的应用
博客专区 > postdep 的博客 > 博客详情
ActiveMQ在C#中的应用
postdep 发表于2年前
ActiveMQ在C#中的应用
  • 发表于 2年前
  • 阅读 82
  • 收藏 0
  • 点赞 0
  • 评论 0

腾讯云 技术升级10大核心产品年终让利>>>   

摘要: 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提供多种语言支持,如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,因此下载最新版即可。

 

[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.         }  

 

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:

[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接收程序

共有 人打赏支持
postdep
粉丝 76
博文 247
码字总数 259349
×
postdep
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: