ActiveMQ在C#中的应用示例分析
脚本之家 / 编程助手:解决程序员“几乎”所有问题!
脚本之家官方知识库 → 点击立即使用
本文实例讲述了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 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | private void StopMonitorThreads() { lock (monitor) { if (monitorStarted.CompareAndSet( true , false )) { AutoResetEvent shutdownEvent = new AutoResetEvent( false ); // Attempt to wait for the Timers to shutdown, but don't wait // forever, if they don't shutdown after two seconds, just quit. this .readCheckTimer.Dispose(shutdownEvent); shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); this .writeCheckTimer.Dispose(shutdownEvent); shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); //WaitOne的定义:public virtual bool WaitOne(TimeSpan timeout,bool exitContext) this .asyncTasks.Shutdown(); this .asyncTasks = null ; this .asyncWriteTask = null ; this .asyncErrorTask = null ; } } } |
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 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | using System; using System.Collections.Generic; using System.Text; using Apache.NMS; using Apache.NMS.ActiveMQ; using System.IO; using System.Xml.Serialization; using System.Runtime.Serialization.Formatters.Binary; namespace Publish { class Program { static void Main( string [] args) { try { //Create the Connection Factory IConnectionFactory factory = new ConnectionFactory( "tcp://localhost:61616/" ); using (IConnection connection = factory.CreateConnection()) { //Create the Session using (ISession session = connection.CreateSession()) { //Create the Producer for the topic/queue IMessageProducer prod = session.CreateProducer( new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic( "testing" )); //Send Messages int i = 0; while (!Console.KeyAvailable) { ITextMessage msg = prod.CreateTextMessage(); msg.Text = i.ToString(); Console.WriteLine( "Sending: " + i.ToString()); prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue); System.Threading.Thread.Sleep(5000); i++; } } } Console.ReadLine(); } catch (System.Exception e) { Console.WriteLine( "{0}" ,e.Message); Console.ReadLine(); } } } } |
consumer:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | using System; using System.Collections.Generic; using System.Text; using Apache.NMS; using Apache.NMS.ActiveMQ; using System.IO; using System.Xml.Serialization; using System.Runtime.Serialization.Formatters.Binary; namespace Subscribe { class Program { static void Main( string [] args) { try { //Create the Connection factory IConnectionFactory factory = new ConnectionFactory( "tcp://localhost:61616/" ); //Create the connection using (IConnection connection = factory.CreateConnection()) { connection.ClientId = "testing listener" ; connection.Start(); //Create the Session using (ISession session = connection.CreateSession()) { //Create the Consumer IMessageConsumer consumer = session.CreateDurableConsumer( new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic( "testing" ), "testing listener" , null , false ); consumer.Listener += new MessageListener(consumer_Listener); Console.ReadLine(); } connection.Stop(); connection.Close(); } } catch (System.Exception e) { Console.WriteLine(e.Message); } } static void consumer_Listener(IMessage message) { try { ITextMessage msg = (ITextMessage)message; Console.WriteLine( "Receive: " + msg.Text); } catch (System.Exception e) { Console.WriteLine(e.Message); } } } } |
程序实现的功能:生产者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 2 3 4 5 | ITextMessage msg = prod.CreateTextMessage(); msg.Text = i.ToString(); msg.Properties.SetString( "myFilter" , "test1" ); Console.WriteLine( "Sending: " + i.ToString()); prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue); |
consumer:
1 2 | //生成consumer时通过参数设置Selector IMessageConsumer consumer = session.CreateConsumer( new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue( "testing" ), "myFilter='test1'" ); |
更多关于C#相关内容感兴趣的读者可查看本站专题:《C#窗体操作技巧汇总》、《C#常见控件用法教程》、《WinForm控件用法总结》、《C#程序设计之线程使用技巧总结》、《C#操作Excel技巧总结》、《C#中XML文件操作技巧汇总》、《C#数据结构与算法教程》、《C#数组操作技巧总结》及《C#面向对象程序设计入门教程》
希望本文所述对大家C#程序设计有所帮助。
微信公众号搜索 “ 脚本之家 ” ,选择关注
程序猿的那些事、送书等活动等着你
相关文章
C# CefSharp 根据输入日期段自动选择日期的操作代码
这篇文章主要介绍了C# CefSharp 根据输入日期段自动选择日期的操作代码,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧2024-01-01C#无法打开计算机“.”上的 IISADMIN 服务的解决方法
在使用c#进行控制IIS服务启动停止的时候,提示:无法打开计算机“.”上的 IISADMIN 服务2015-01-01
最新评论