首页 / C# / ActiveMQ在C#中的应用
ActiveMQ在C#中的应用
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了ActiveMQ在C#中的应用，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含6956字，纯文字阅读大概需要10分钟。
内容图文
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
private
void
StopMonitorThreads()
2
{
3
lock
(monitor)
4
{
5
if(monitorStarted.CompareAndSet(true, false))
6 {
7 AutoResetEvent shutdownEvent = new AutoResetEvent(false);
8// Attempt to wait for the Timers to shutdown, but don‘t wait
9// forever, if they don‘t shutdown after two seconds, just quit. 10this.readCheckTimer.Dispose(shutdownEvent);
11 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
12this.writeCheckTimer.Dispose(shutdownEvent);
13 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
14//WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 15this.asyncTasks.Shutdown();
16this.asyncTasks = null;
17this.asyncWriteTask = null;
18this.asyncErrorTask = null;
19 }
20 }
21 }
22privatevoid StopMonitorThreads()
23 {
24lock(monitor)
25 {
26if(monitorStarted.CompareAndSet(true, false))
27 {
28 AutoResetEvent shutdownEvent = new AutoResetEvent(false);
2930// Attempt to wait for the Timers to shutdown, but don‘t wait
31// forever, if they don‘t shutdown after two seconds, just quit. 32this.readCheckTimer.Dispose(shutdownEvent);
33 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
34this.writeCheckTimer.Dispose(shutdownEvent);
35 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
36//WaitOne的定义：public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 37this.asyncTasks.Shutdown();
38this.asyncTasks = null;
39this.asyncWriteTask = null;
40this.asyncErrorTask = null;
41 }
42 }
43 }
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
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
namespace
Publish
10
{
11
class
Program
12
{
13
static
void Main(string[] args)
14 {
15try16 {
17//Create the Connection Factory 18 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
19using (IConnection connection = factory.CreateConnection())
20 {
21//Create the Session 22using (ISession session = connection.CreateSession())
23 {
24//Create the Producer for the topic/queue 25 IMessageProducer prod = session.CreateProducer(
26new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"));
27//Send Messages 28int i = 0;
29while (!Console.KeyAvailable)
30 {
31 ITextMessage msg = prod.CreateTextMessage();
32 msg.Text = i.ToString();
33 Console.WriteLine("Sending: " + i.ToString());
34 prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
35 System.Threading.Thread.Sleep(5000);
36 i++;
37 }
38 }
39 }
40 Console.ReadLine();
41 }
42catch (System.Exception e)
43 {
44 Console.WriteLine("{0}",e.Message);
45 Console.ReadLine();
46 }
47 }
48 }
49 }
consumer:
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
namespace
Subscribe
10
{
11
class
Program
12
{
13
static
void Main(string[] args)
14 {
15try16 {
17//Create the Connection factory 18 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/");
19//Create the connection 20using (IConnection connection = factory.CreateConnection())
21 {
22 connection.ClientId = "testing listener";
23 connection.Start();
24//Create the Session 25using (ISession session = connection.CreateSession())
26 {
27//Create the Consumer 28 IMessageConsumer consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"), "testing listener", null, false);
29 consumer.Listener += new MessageListener(consumer_Listener);
30 Console.ReadLine();
31 }
32 connection.Stop();
33 connection.Close();
34 }
35 }
36catch (System.Exception e)
37 {
38 Console.WriteLine(e.Message);
39 }
40 }
41staticvoid consumer_Listener(IMessage message)
42 {
43try44 {
45 ITextMessage msg = (ITextMessage)message;
46 Console.WriteLine("Receive: " + msg.Text);
47 }
48catch (System.Exception e)
49 {
50 Console.WriteLine(e.Message);
51 }
52 }
53 }
54 }
程序实现的功能：生产者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
public
void
SetProperties()
2
{
3 ITextMessage msg = prod.CreateTextMessage();
4 msg.Text = i.ToString();
5 msg.Properties.SetString("myFilter", "test1");
6 Console.WriteLine("Sending: " + i.ToString());
7 prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
8 ITextMessage msg = prod.CreateTextMessage();
9 msg.Text = i.ToString();
10 msg.Properties.SetString("myFilter", "test1");
11 Console.WriteLine("Sending: " + i.ToString());
12 prod.Send(msg, Apache.NMS.MsgDeliveryMode.NonPersistent, Apache.NMS.MsgPriority.Normal, TimeSpan.MinValue);
1314 }
consumer：
1
public
void
SetSelector()
2
{
3
//
生成consumer时通过参数设置Selector
4 IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter=‘test1‘");
5//生成consumer时通过参数设置Selector 6 IMessageConsumer consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"), "myFilter=‘test1‘");
7 }
该文章来自博客园：http://www.cnblogs.com/guthing/archive/2010/06/17/1759333.html
经过简单的整理，感谢guthing
ActiveMQ 其他文章推荐：http://blog.csdn.net/lee353086/article/details/6819123
原文：http://www.cnblogs.com/pudefu/p/7562787.html
内容总结
以上是互联网集市为您收集整理的ActiveMQ在C#中的应用全部内容，希望文章能够帮你解决ActiveMQ在C#中的应用所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。