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，因此下载最新版即可。

view plaincopy to clipboardprint?

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：

view plaincopy to clipboardprint?

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:

view plaincopy to clipboardprint?

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"),

查看更多 下一篇