ActiveMQ在C#中的应用示例分析
本文实例讲述了ActiveMQ在C#中的应用。分享给大家供大家参考，具体如下：
ActiveMQ是个好东东，不必多说。ActiveMQ提供多种语言支持，如Java,C,C++,C#,Ruby,Perl,Python,PHP等。由于我在windows下开发GUI，比较关心C++和C#，其中C#的ActiveMQ很简单，Apache提供NMS（.NetMessagingService）支持.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，因此下载最新版即可。
privatevoidStopMonitorThreads()
{
lock(monitor)
{
if(monitorStarted.CompareAndSet(true,false))
{
AutoResetEventshutdownEvent=newAutoResetEvent(false);
//AttempttowaitfortheTimerstoshutdown,butdon'twait
//forever,iftheydon'tshutdownaftertwoseconds,justquit.
this.readCheckTimer.Dispose(shutdownEvent);
shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
this.writeCheckTimer.Dispose(shutdownEvent);
shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000));
//WaitOne的定义：publicvirtualboolWaitOne(TimeSpantimeout,boolexitContext)
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：
usingSystem;
usingSystem.Collections.Generic;
usingSystem.Text;
usingApache.NMS;
usingApache.NMS.ActiveMQ;
usingSystem.IO;
usingSystem.Xml.Serialization;
usingSystem.Runtime.Serialization.Formatters.Binary;
namespacePublish
{
classProgram
{
staticvoidMain(string[]args)
{
try
{
//CreatetheConnectionFactory
IConnectionFactoryfactory=newConnectionFactory("tcp://localhost:61616/");
using(IConnectionconnection=factory.CreateConnection())
{
//CreatetheSession
using(ISessionsession=connection.CreateSession())
{
//CreatetheProducerforthetopic/queue
IMessageProducerprod=session.CreateProducer(
newApache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"));
//SendMessages
inti=0;
while(!Console.KeyAvailable)
{
ITextMessagemsg=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.Exceptione)
{
Console.WriteLine("{0}",e.Message);
Console.ReadLine();
}
}
}
}
consumer:
usingSystem;
usingSystem.Collections.Generic;
usingSystem.Text;
usingApache.NMS;
usingApache.NMS.ActiveMQ;
usingSystem.IO;
usingSystem.Xml.Serialization;
usingSystem.Runtime.Serialization.Formatters.Binary;
namespaceSubscribe
{
classProgram
{
staticvoidMain(string[]args)
{
try
{
//CreatetheConnectionfactory
IConnectionFactoryfactory=newConnectionFactory("tcp://localhost:61616/");
//Createtheconnection
using(IConnectionconnection=factory.CreateConnection())
{
connection.ClientId="testinglistener";
connection.Start();
//CreatetheSession
using(ISessionsession=connection.CreateSession())
{
//CreatetheConsumer
IMessageConsumerconsumer=session.CreateDurableConsumer(newApache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing"),"testinglistener",null,false);
consumer.Listener+=newMessageListener(consumer_Listener);
Console.ReadLine();
}
connection.Stop();
connection.Close();
}
}
catch(System.Exceptione)
{
Console.WriteLine(e.Message);
}
}
staticvoidconsumer_Listener(IMessagemessage)
{
try
{
ITextMessagemsg=(ITextMessage)message;
Console.WriteLine("Receive:"+msg.Text);
}
catch(System.Exceptione)
{
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-->consumerA
msg2-->consumerB
msg3-->consumerC
msg4-->consumerA
msg5-->consumerB
msg6-->consumerC
特殊情况是指：ActiveMQ支持过滤机制，即生产者可以设置消息的属性（Properties），该属性与消费者端的Selector对应，只有消费者设置的selector与消息的Properties匹配，消息才会发给该消费者。Topic和Queue都支持Selector。
Properties和Selector该如何设置呢？请看如下代码：
producer：
ITextMessagemsg=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：
//生成consumer时通过参数设置Selector
IMessageConsumerconsumer=session.CreateConsumer(newApache.NMS.ActiveMQ.Commands.ActiveMQQueue("testing"),"myFilter='test1'");
更多关于C#相关内容感兴趣的读者可查看本站专题：《C#窗体操作技巧汇总》、《C#常见控件用法教程》、《WinForm控件用法总结》、《C#程序设计之线程使用技巧总结》、《C#操作Excel技巧总结》、《C#中XML文件操作技巧汇总》、《C#数据结构与算法教程》、《C#数组操作技巧总结》及《C#面向对象程序设计入门教程》
希望本文所述对大家C#程序设计有所帮助。