文档章节

ActiveMQ在C#中的应用

postdep
 postdep
发布于 2015/08/24 22:15
字数 1231
阅读 94
收藏 1
点赞 0
评论 0

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
博文 252
码字总数 261672
作品 0
武汉
高级程序员
ActiveMQ集群方案(下)

版权声明:欢迎转载,但是看在我辛勤劳动的份上,请注明来源:http://blog.csdn.net/yinwenjie(未经允许严禁用于商业用途!) https://blog.csdn.net/yinwenjie/article/details/51205822 目...

yunlielai ⋅ 04/15 ⋅ 0

ActiveMQ初探(1)——介绍与基本使用

一、ActiveMQ 1.1 什么是ActiveMQ 是Apache出品,最流行的,能力强劲的。ActiveMQ是一个完全支持和规范的 实现,尽管规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊...

yuanlaijike ⋅ 04/15 ⋅ 0

ActiveMQ集群方案(上)

版权声明:欢迎转载,但是看在我辛勤劳动的份上,请注明来源:http://blog.csdn.net/yinwenjie(未经允许严禁用于商业用途!) https://blog.csdn.net/yinwenjie/article/details/51124749 目...

yunlielai ⋅ 04/15 ⋅ 0

爬虫架构 | 消息队列应用场景及ActiveMQ、RabbitMQ、RocketMQ、Kafka对比

前言:在之前的业务中,使用了Kafka和RabbitMQ两种消息队列,这篇文章来做一个总结。 消息队列中间件是分布式系统中重要的组件,主要实现异步消息,应用解耦,流量削峰及消息通讯等功能。 下...

小怪聊职场 ⋅ 04/26 ⋅ 0

ActiveMQ初探(2)——与Spring整合

在上一章:ActiveMQ(1)——介绍与基本使用中,我们通过测试方法熟悉了ActiveMQ的流程,本章将ActiveMQ与Spring进行整合。 与Spring整合,除了原本的外,还需导入和包,如果Spring为5.0+,需...

yuanlaijike ⋅ 05/08 ⋅ 0

消息队列篇—详谈ActiveMQ消息队列模式的分析及使用

消息队列(Message Queue)是分布式系统中重要的组件,通用使用场景可以简单地描述为当不需要立即获得结果,但是并发量需控制时就需要使用消息队列。消息列队有两种消息模式,一种是点对点的...

afreon ⋅ 04/22 ⋅ 0

activeMQ发送邮件小例子

先安装apache-activemq-5.11.1 ,然后启动activemq.bat 启动activeMQ的控制台:http://localhost:8161/admin/,展示图如下: 解释:Number Of Pending Messages :等待消费的消息,这个是当前...

zhouyuzhoujing ⋅ 05/07 ⋅ 0

ActiveMQ初探(3)——与SpringBoot整合

在上一章:ActiveMQ(2)——与Spring整合中,我们已经完成了ActiveMQ与Spring进行整合,本章将进行与SpringBoot整合。 发送即时消息 Step1:添加依赖包: 这里我们不使用网上大多数的,因为...

yuanlaijike ⋅ 05/08 ⋅ 0

ActiveMQ RabbitMQ KafKa对比

前言: ActiveMQ和 RabbitMq 以及Kafka在之前的项目中都有陆续使用过,当然对于三者没有进行过具体的对比,以下摘抄了一些网上关于这三者的对比情况,我自己看过之后感觉还 是可以的,比较清...

xiaomin0322 ⋅ 05/11 ⋅ 0

ActiveMq笔记2-消息持久化

为了避免意外宕机以后丢失信息,需要做到重启后可以恢复消息队列,消息系统一般都会采用持久化机制。 ActiveMQ的消息持久化机制有JDBC,AMQ,KahaDB和LevelDB, 无论使用哪种持久化方式,消息...

狂小白 ⋅ 02/25 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

zblog2.3版本的asp系统是否可以超越卢松松博客的流量[图]

最近访问zblog官网,发现zlbog-asp2.3版本已经进入测试阶段了,虽然正式版还没有发布,想必也不久了。那么作为aps纵横江湖十多年的今天,blog2.2版本应该已经成熟了,为什么还要发布这个2.3...

原创小博客 ⋅ 今天 ⋅ 0

聊聊spring cloud的HystrixCircuitBreakerConfiguration

序 本文主要研究一下spring cloud的HystrixCircuitBreakerConfiguration HystrixCircuitBreakerConfiguration spring-cloud-netflix-core-2.0.0.RELEASE-sources.jar!/org/springframework/......

go4it ⋅ 今天 ⋅ 0

二分查找

二分查找,也称折半查找、二分搜索,是一种在有序数组中查找某一特定元素的搜索算法。搜素过程从数组的中间元素开始,如果中间元素正好是要查找的元素,则搜素过程结束;如果某一特定元素大于...

人觉非常君 ⋅ 今天 ⋅ 0

VS中使用X64汇编

需要注意的是,在X86项目中,可以使用__asm{}来嵌入汇编代码,但是在X64项目中,再也不能使用__asm{}来编写嵌入式汇编程序了,必须使用专门的.asm汇编文件来编写相应的汇编代码,然后在其它地...

simpower ⋅ 今天 ⋅ 0

ThreadPoolExecutor

ThreadPoolExecutor public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, ......

4rnold ⋅ 昨天 ⋅ 0

Java正无穷大、负无穷大以及NaN

问题来源:用Java代码写了一个计算公式,包含除法和对数和取反,在页面上出现了-infinity,不知道这是什么问题,网上找答案才明白意思是负的无穷大。 思考:为什么会出现这种情况呢?这是哪里...

young_chen ⋅ 昨天 ⋅ 0

前台对中文编码,后台解码

前台:encodeURI(sbzt) 后台:String param = URLDecoder.decode(sbzt,"UTF-8");

west_coast ⋅ 昨天 ⋅ 0

实验楼—MySQL基础课程-挑战3实验报告

按照文档要求创建数据库 sudo sercice mysql startwget http://labfile.oss.aliyuncs.com/courses/9/createdb2.sqlvim /home/shiyanlou/createdb2.sql#查看下数据库代码 代码创建了grade......

zhangjin7 ⋅ 昨天 ⋅ 0

一起读书《深入浅出nodejs》-node模块机制

node 模块机制 前言 说到node,就不免得提到JavaScript。JavaScript自诞生以来,经历了工具类库、组件库、前端框架、前端应用的变迁。通过无数开发人员的努力,JavaScript不断被类聚和抽象,...

小草先森 ⋅ 昨天 ⋅ 0

Java桌球小游戏

其实算不上一个游戏,就是两张图片,不停的重画,改变ball图片的位置。一个左右直线碰撞的,一个有角度碰撞的。 左右直线碰撞 package com.bjsxt.test;import javax.swing.*;import j...

森林之下 ⋅ 昨天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部