文档章节

ActiveMQ在C#中的应用

postdep
 postdep
发布于 2015/08/24 22:15
字数 1231
阅读 101
收藏 1

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

本文转载自:http://blog.csdn.net/bodybo/article/details/5647968

共有 人打赏支持
postdep

postdep

粉丝 80
博文 255
码字总数 261672
作品 0
武汉
高级程序员
JMS配置说明-----activeMQ-5.6

1 简介 activeMQ是一个完全支持JMS1.1 和J2EE规范的JMS Provider实现; 尽管规范出台已经是很久的事情了,但JMS在当今的J2EE应用中仍然扮演着特殊的地位; 特性列表 多种语言和协议编写客户端...

次渠龙哥
06/26
0
0
ActiveMQ安装配置和使用简例

本文作者:Zhang Phil 原文链接:http://blog.csdn.net/zhangphil/article/details/48173665 ActiveMQ安装配置和使用简例 ActiveMQ是一套JMS(Java Message Service)开源消息服务实现的组件...

开开心心过
2015/09/02
0
0
Linux 安装ActiveMQ(使用Mac远程访问)

阅读本文需要安装JDK 一 ActiveMQ简介 activemq是用java语言编写的一款开源消息总线 activemq是apache出品 activemq消息的传递有两种类型 一种是点对点(即一个生产者和一个消费者一一对应) 另...

梦三
07/15
0
0
.Net平台下ActiveMQ入门实例

1.ActiveMQ简介 先分析这么一个场景:当我们在网站上购物时,必须经过,下订单、发票创建、付款处理、订单履行、航运等。但是,当用户下单后,立即跳转到“感谢那您的订单” 页面。不仅如此,...

postdep
2015/08/24
0
0
Win7环境下安装ActiveMQ

参考ActiveMQ官方文档:http://activemq.apache.org/getting-started.html 安装ActiveMQ 近来要学习JMS,在网上查了些资料,发现ActiveMQ是比较流行的JMS开源框架,决定使用ActiveMQ来学习J...

纠结名字
2015/08/09
0
0

没有更多内容

加载失败,请刷新页面

加载更多

cacti监控安装

cacti是用PHP实现的一个软件,它用snmp服务获取数据,然后用rrdtool存储和更新数据,并生成图表展示。比较适合用于交换机、路由器的网络监控,插件众多,可图示化显示网络状况。 cacti官方推...

hiwill
5分钟前
0
0
shell特殊符号、cut、sort、uniq、wc、tee、tr、split命令

10月15日任务 8.10 shell特殊符号cut命令 8.11 sort_wc_uniq命令 8.12 tee_tr_split命令 8.13 shell特殊符号下 cut 命令 cut作用:截取字符串 用法如下:cat /etc/passwd |head -2 |cut -d ...

hhpuppy
34分钟前
1
0
Springboot实现filter拦截token验证和跨域

背景 web验证授权合法的一般分为下面几种 1使用session作为验证合法用户访问的验证方式 使用自己实现的token 使用OCA标准 在使用API接口授权验证时,token是自定义的方式实现起来不需要引入其...

funnymin
今天
3
0
linux使用ntfs-3g操作ntfs格式硬盘

Linux内核目前只支持对微软NTFS文件系统的读取。 NTFS-3G 是微软 NTFS 文件系统的一个开源实现,同时支持读和写。NTFS-3G 开发者使用 FUSE 文件系统来辅助开发,同时对可移植性有益。 安装 ...

linuxprobe16
今天
1
0
kubeadm部署kubernetes集群

一、环境要求 这里使用RHEL7.5 master、etcd:192.168.10.101,主机名:master node1:192.168.10.103,主机名:node1 node2:192.168.10.104,主机名:node2 所有机子能基于主机名通信,编辑...

人在艹木中
今天
14
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部