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. 10 this.readCheckTimer.Dispose(shutdownEvent); 11 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 12 this.writeCheckTimer.Dispose(shutdownEvent); 13 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 14 //WaitOne的定义:public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 15 this.asyncTasks.Shutdown(); 16 this.asyncTasks = null; 17 this.asyncWriteTask = null; 18 this.asyncErrorTask = null; 19 } 20 } 21 }
22 private void StopMonitorThreads() 23 { 24 lock(monitor) 25 { 26 if(monitorStarted.CompareAndSet(true, false)) 27 { 28 AutoResetEvent shutdownEvent = new AutoResetEvent(false); 29 30 // Attempt to wait for the Timers to shutdown, but don't wait 31 // forever, if they don't shutdown after two seconds, just quit. 32 this.readCheckTimer.Dispose(shutdownEvent); 33 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 34 this.writeCheckTimer.Dispose(shutdownEvent); 35 shutdownEvent.WaitOne(TimeSpan.FromMilliseconds(2000)); 36 //WaitOne的定义:public virtual bool WaitOne(TimeSpan timeout,bool exitContext) 37 this.asyncTasks.Shutdown(); 38 this.asyncTasks = null; 39 this.asyncWriteTask = null; 40 this.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 { 15 try 16 { 17 //Create the Connection Factory 18 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/"); 19 using (IConnection connection = factory.CreateConnection()) 20 { 21 //Create the Session 22 using (ISession session = connection.CreateSession()) 23 { 24 //Create the Producer for the topic/queue 25 IMessageProducer prod = session.CreateProducer( 26 new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("testing")); 27 //Send Messages 28 int i = 0; 29 while (!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 } 42 catch (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 { 15 try 16 { 17 //Create the Connection factory 18 IConnectionFactory factory = new ConnectionFactory("tcp://localhost:61616/"); 19 //Create the connection 20 using (IConnection connection = factory.CreateConnection()) 21 { 22 connection.ClientId = "testing listener"; 23 connection.Start(); 24 //Create the Session 25 using (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 } 36 catch (System.Exception e) 37 { 38 Console.WriteLine(e.Message); 39 } 40 } 41 static void consumer_Listener(IMessage message) 42 { 43 try 44 { 45 ITextMessage msg = (ITextMessage)message; 46 Console.WriteLine("Receive: " + msg.Text); 47 } 48 catch (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); 13 14 }
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