首页 > 代码库 > ActiveMQ
ActiveMQ
前言
MQ——Message Queue,中文翻译为“消息队列”，维基百科上的这样描述：
消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式，软件的贮列用来处理一系列的输入，通常是来自使用者。消息队列提供了异步的通信协议，每一个贮列中的纪录包含详细说明的资料，包含发生的时间，输入装置的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不需要同时与消息队列互交。消息会保存在队列中，直到接收者取回它。
实际应用中有多种MQ，包括MSMQ,Active MQ,Jboss MQ等.本文简单介绍.net环境下Active MQ入门。
准备工作
1.下载Active MQ，地址：http://activemq.apache.org/download.html(我下的是ActiveMQ 5.7.0 Release),下载完成后解压缩即可。ActiveMQ是一个开源的JMS服务器；
2.下载NMS，地址：http://activemq.apache.org/nms/download.html。网站提供源码下载，可以根据需要编译成指定.net framework版本。如果下载的是release版本，直接将具体.net版本目录下Apache.NMS.ActiveMQ.dll和Apache.NMS.dll拷贝出来。
Hello ActiveMQ
1.新建空解决方案：HelloActiveMQ.sln
2.在当前解决方案中分别新建两个控制台程序：ConsoleCustomer.csproj（消费者）和ConsoleProcucer.csproj（生产者）。生产者将消息发送给ActiveMQ,然后ActiveMQ将消息推送给符合条件的消费者；
3.在ConsoleCustomer.csproj（消费者）和ConsoleProcucer.csproj（生产者）两个项目中均添加对Apache.NMS.ActiveMQ.dll和Apache.NMS.dll的引用；
4.在ConsoleProcucer项目的program.cs中添加如下代码（代码中已经又注释）：
class Program { static IConnectionFactory _factory = null; static IConnection _connection = null; static ITextMessage _message = null; static void Main(string[] args) { //创建工厂 _factory = new ConnectionFactory("tcp://localhost:61616/"); try { //创建连接 using (_connection = _factory.CreateConnection()) { //创建会话 using (ISession session = _connection.CreateSession()) { //创建一个主题 IDestination destination = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic"); //创建生产者 IMessageProducer producer = session.CreateProducer(destination); Console.WriteLine("Please enter any key to continue! "); Console.ReadKey(); Console.WriteLine("Sending: "); //创建一个文本消息 _message = producer.CreateTextMessage("Hello AcitveMQ...."); //发送消息 producer.Send(_message, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue); } } } catch (Exception ex) { Console.WriteLine(ex.ToString()); } Console.ReadLine(); } }
5.在ConsoleCustomer项目的program.cs中添加如下代码：
class Program { static IConnectionFactory _factory = null; static void Main(string[] args) { try { //创建连接工厂 _factory = new ConnectionFactory("tcp://localhost:61616/"); //创建连接 using (IConnection conn = _factory.CreateConnection()) { //设置客户端ID conn.ClientId = "Customer"; conn.Start(); //创建会话 using (ISession session = conn.CreateSession()) { //创建主题 var topic = new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic("topic"); //创建消费者 IMessageConsumer consumer = session.CreateDurableConsumer(topic, "Customer", null, false); //注册监听事件 consumer.Listener += new MessageListener(consumer_Listener); //这句代码非常重要， //这里没有read方法，Session会话会被关闭，那么消费者将监听不到生产者的消息 Console.Read(); } //关闭连接 conn.Stop(); conn.Close(); } } catch (Exception ex) { Console.Write(ex.ToString()); } } /// <summary> /// 消费监听事件 /// </summary> /// <param name="message"></param> static void consumer_Listener(IMessage message) { ITextMessage msg = (ITextMessage)message; Console.WriteLine("Receive: " + msg.Text); }
上述代码生成成功后，一个ActiveMQ的入门程序已经完成，接下来便是测试了。
测试
1.启动之前解压过的ActiveMQ程序，在apache-activemq-5.7.0\bin目录下运行activemq.bat批处理文件。运行截图如下：
2.分别运行之前生成的ConsoleProcucer.exe.和ConsoleCustomer.exe，在ConsoleProcucer.exe中根据提示输入任意键。我们便可以检测ConsoleCustomer.exe有没有接收到我们发送的“Hello ActiveMQ”文本。结果如下：
ConsoleProcucer.exe截图
ConsoleCustomer.exe截图
通过验证我们发现通过ActiveMQ发送消息成功。
总结：
1.ConnectionFactory 是连接工厂，负责创建Connection，Connection 负责创建 Session，Session 创建 MessageProducer（用来发消息） 和 MessageConsumer（用来接收消息），Destination 是消息的目的地；
2.Connection一定要调用Start()，才能接收消息。关闭Connection时先调用Stop()，然后再调用Close()；
3.Session在接收消息之前不能关闭，否则监听不到消息；
4.Connection,Session不使用时应该及时关闭，避免内存泄露;
5.ActiveMQ支持topic（主题，广播消息）和Queue（队列，点对点消息）。
注：源码中另外提供了winform实现。需要注意的是winform中消息监听事件在接受到消息后，由于当前不是处于UI线程，是不能直接操作Winform UI的，需要使用Invoke，详情请参照源码。
点击下载源代码
ActiveMQ