ActiveMq C#客户端 消息队列的使用（存和取）

1、准备工具

 

1. VS2013
2. Apache.NMS.ActiveMQ-1.7.2-bin.zip
3. apache-activemq-5.14.0-bin.zip

 

2、开始项目

 

VS2013新建一个C#控制台应用程序，项目中添加两个dll引用，一个是D:Apache.NMS.ActiveMQ-1.7.2-binlibApache.NMS et-4.0目录下的Apache.NMS.dll，另一个是D:Apache.NMS.ActiveMQ-1.7.2-binuild et-4.0debug目录下的Apache.NMS.ActiveMQ.dll。

新建一个类，MyActiveMq.cs，用于对activemq消息队列接口的封装，实现如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

using Apache.NMS;
using Apache.NMS.ActiveMQ;

namespace NmsProducerClasses
{
    public class MyActiveMq
    {
        private IConnectionFactory factory;
        private IConnection connection;
        private ISession session;
        private IMessageProducer prod;
        private IMessageConsumer consumer;
        private ITextMessage msg;

        private bool isTopic = false;
        private bool hasSelector = false;
        private const string ClientID = "clientid";
        private const string Selector = "filter='demo'";
        private bool sendSuccess = true;
        private bool receiveSuccess = true;

        public MyActiveMq(bool isLocalMachine, string remoteAddress)
        {
            try
            {
                //初始化工厂
                if (isLocalMachine)
                {
                    factory = new ConnectionFactory("tcp://localhost:61616/");
                }
                else
                {
                    factory = new ConnectionFactory("tcp://" + remoteAddress + ":61616/"); //写tcp://192.168.1.111:61616的形式连接其他服务器上的ActiveMQ服务器
                }
                //通过工厂建立连接
                connection = factory.CreateConnection();
                connection.ClientId = ClientID;
                connection.Start();
                //通过连接创建Session会话
                session = connection.CreateSession();
            }
            catch (System.Exception e)
            {
                sendSuccess = false;
                receiveSuccess = false;
                Console.WriteLine("Exception:{0}", e.Message);
                Console.ReadLine();
                throw e;
            }
            Console.WriteLine("Begin connection...");
        }


        ~MyActiveMq()
        {
            //this.ShutDown();
        }

        /// <summary>
        /// 初始化
        /// </summary>
        /// <param name="topic">选择是否是Topic</param>
        /// <param name="name">队列名</param>
        /// <param name="selector">是否设置过滤</param>
        public bool InitQueueOrTopic(bool topic, string name, bool selector = false)
        {
            try
            {
                //通过会话创建生产者、消费者
                if (topic)
                {
                    prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic(name));
                    if (selector)
                    {
                        consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic(name), ClientID, Selector, false);
                        hasSelector = true;
                    }
                    else
                    {
                        consumer = session.CreateDurableConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQTopic(name), ClientID, null, false);
                        hasSelector = false;
                    }
                    isTopic = true;
                }
                else
                {
                    prod = session.CreateProducer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue(name));
                    if (selector)
                    {
                        consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue(name), Selector);
                        hasSelector = true;
                    }
                    else
                    {
                        consumer = session.CreateConsumer(new Apache.NMS.ActiveMQ.Commands.ActiveMQQueue(name));
                        hasSelector = false;
                    }
                    isTopic = false;
                }
                //创建一个发送的消息对象
                msg = prod.CreateTextMessage();
            }
            catch (System.Exception e)
            {
                sendSuccess = false;
                receiveSuccess = false;
                Console.WriteLine("Exception:{0}", e.Message);
                Console.ReadLine();
                throw e;
            }

            return sendSuccess;
        }


        public bool SendMessage(string message, string msgId = "defult", MsgPriority priority = MsgPriority.Normal)
        {
            if (prod == null)
            {
                sendSuccess = false;
                Console.WriteLine("call InitQueueOrTopic() first!!");
                return false;
            }

            Console.WriteLine("Begin send messages...");

            //给这个对象赋实际的消息
            msg.NMSCorrelationID = msgId;
            msg.Properties["MyID"] = msgId;
            msg.NMSMessageId = msgId;
            msg.Text = message;
            Console.WriteLine(message);

            if (isTopic)
            {
                sendSuccess = ProducerSubcriber(message, priority);
            }
            else
            {
                sendSuccess = P2P(message, priority);
            }

            return sendSuccess;
        }


        public string GetMessage()
        {
            if (prod == null)
            {
                Console.WriteLine("call InitQueueOrTopic() first!!");
                return null;
            }

            Console.WriteLine("Begin receive messages...");
            ITextMessage revMessage = null;
            try
            {
                //同步阻塞10ms,没消息就直接返回null,注意此处时间不能设太短，否则还没取到消息就直接返回null了！！！
                revMessage = consumer.Receive(new TimeSpan(TimeSpan.TicksPerMillisecond *10)) as ITextMessage;
            }
            catch (System.Exception e)
            {
                receiveSuccess = false;
                Console.WriteLine("Exception:{0}", e.Message);
                Console.ReadLine();
                throw e;
            }

            if (revMessage == null)
            {
                Console.WriteLine("No message received!");
                return null;
            }
            else
            {
                Console.WriteLine("Received message with Correlation ID: " + revMessage.NMSCorrelationID);
                //Console.WriteLine("Received message with Properties'ID: " + revMessage.Properties["MyID"]);
                Console.WriteLine("Received message with text: " + revMessage.Text);
            }

            return revMessage.Text;
        }

        //P2P模式，一个生产者对应一个消费者
        private bool P2P(string message, MsgPriority priority)
        {
            try
            {
                if (hasSelector)
                {
                    //设置消息对象的属性，这个很重要，是Queue的过滤条件，也是P2P消息的唯一指定属性
                    msg.Properties.SetString("filter", "demo");  //P2P模式
                }
                prod.Priority = priority;
                //设置持久化
                prod.DeliveryMode = MsgDeliveryMode.Persistent;
                //生产者把消息发送出去，几个枚举参数MsgDeliveryMode是否持久化，MsgPriority消息优先级别，存活时间，当然还有其他重载
                prod.Send(msg, MsgDeliveryMode.Persistent, priority, TimeSpan.MinValue);
            }
            catch (System.Exception e)
            {
                sendSuccess = false;
                Console.WriteLine("Exception:{0}", e.Message);
                Console.ReadLine();
                throw e;
            }

            return sendSuccess;
        }


        //发布订阅模式，一个生产者多个消费者
        private bool ProducerSubcriber(string message, MsgPriority priority)
        {
            try
            {
                prod.Priority = priority;
                //设置持久化,如果DeliveryMode没有设置或者设置为NON_PERSISTENT，那么重启MQ之后消息就会丢失
                prod.DeliveryMode = MsgDeliveryMode.Persistent;
                prod.Send(msg, Apache.NMS.MsgDeliveryMode.Persistent, priority, TimeSpan.MinValue);
                //System.Threading.Thread.Sleep(1000);
            }
            catch (System.Exception e)
            {
                sendSuccess = false;
                Console.WriteLine("Exception:{0}", e.Message);
                Console.ReadLine();
                throw e;
            }

            return sendSuccess;
        }


        public void ShutDown()
        {
            Console.WriteLine("Close connection and session...");
            session.Close();
            connection.Close();
        }
    }
}

Program.cs代码如下：

 

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.IO;
using System.Threading;

namespace NmsProducerClasses
{
    class Program
    {
        static void Main(string[] args)
        {
            MyActiveMq mymq = new MyActiveMq(isLocalMachine: true, remoteAddress: "");

            mymq.InitQueueOrTopic(topic: false, name: "myqueue", selector: false);
            //mymq.InitQueueOrTopic(topic: false, name: "seletorqueue", selector: true);
            //mymq.InitQueueOrTopic(topic: true, name: "noselectortopic", selector: false);
            //mymq.InitQueueOrTopic(topic: true, name: "selectortopic", selector: true);

            //The full range of priority values (0-9) are supported by the JDBC message store. For KahaDB three priority categories are supported, Low (< 4), Default (= 4) and High (> 4).
            User myuser0 = new User("0000", "Lowest", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser0), "newid", priority: Apache.NMS.MsgPriority.Lowest);
            User myuser1 = new User("1111", "AboveLow", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser1), "newid", priority: Apache.NMS.MsgPriority.AboveLow);
            User myuser2 = new User("2222", "AboveNormal", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser2), "newid", priority: Apache.NMS.MsgPriority.AboveNormal);
            User myuser3 = new User("0000", "BelowNormal", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser3), "newid", priority: Apache.NMS.MsgPriority.BelowNormal);
            User myuser4 = new User("1111", "High", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser4), "newid", priority: Apache.NMS.MsgPriority.High);
            User myuser5 = new User("2222", "Highest", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser5), "newid", priority: Apache.NMS.MsgPriority.Highest);
            User myuser6 = new User("0000", "Low", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser6), "newid", priority: Apache.NMS.MsgPriority.Low);
            User myuser7 = new User("1111", "Normal", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser7), "newid", priority: Apache.NMS.MsgPriority.Normal);
            User myuser8 = new User("2222", "VeryHigh", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser8), "newid", priority: Apache.NMS.MsgPriority.VeryHigh);
            User myuser9 = new User("2222", "VeryLow", "img/p.jpg");
            mymq.SendMessage(JsonUtil.ObjectToJson(myuser8), "newid", priority: Apache.NMS.MsgPriority.VeryLow);

            int num = 20;
            while (num-- > 0)
            {
                mymq.GetMessage();
                //Thread.Sleep(1000);
            }
            mymq.ShutDown();


            //XML测试
            //string xml = XmlTest.ObjToXml();
            //Console.WriteLine("ObjToXml: {0}", xml);

            //Json测试
            //User u = new User() { Id="88", Imgurl="img/88.jpg", Name="haha88"};
            //string jsonstr = JsonUtil.ObjectToJson(u);
            //Console.WriteLine(jsonstr);

        }

    }

3、测试

首先，需要启动消息队列，具体启动及测试消息队列步骤可见这边：点击打开链接

然后，运行项目，运行结果如下：

 

 

 

 

4、优先级

priority并不能决定消息传送的严格消息，具体原因可见

http://activemq.apache.org/how-can-i-support-priority-queues.html

http://shift-alt-ctrl.iteye.com/blog/2034440 

优先级设置：

在D:apache-activemq-5.14.0conf目录的activemq.xml配置文件中，找到<destinationPolicy>标签，在其中的<policyEntries>标签下添加

<policyEntry queue=">"  producerFlowControl="false" prioritizedMessages="true" useCache="false" expireMessagesPeriod="0" queuePrefetch="1" />
<policyEntry queue=">" strictOrderDispatch="false" />
<policyEntry queue=">" >
              <pendingMessageLimitStrategy>
                  <constantPendingMessageLimitStrategy limit="0"/>
              </pendingMessageLimitStrategy>
              <messageEvictionStrategy>
                  <oldestMessageWithLowestPriorityEvictionStrategy/>
              </messageEvictionStrategy>
</policyEntry>

 

配置完成后，需要重启activemq

 

 

5、远程登录监控

要实现远程监控服务器消息队列，需要先进行配置。

配置方法：在D:apache-activemq-5.14.0conf目录的jetty.xml配置文件中，把133开始的那段注释去掉即可。