首页 > 代码库 > 微软MSMQ消息队列的使用

微软MSMQ消息队列的使用

2024-07-07 22:57:39   227人阅读

首先在windows系统中安装MSMQ

一、MSMQ交互

开发基于消息的应用程序从队列开始。MSMQ包含四种队列类型：

  • 外发队列：消息发送到目的地之前，用它来临时存储消息。
  • 公共队列：在主动目录中公布。整个网络各种服务器上的应用程序能够通过主动目录找到并应用公共队列。
  • 私有队列：这些是本地服务器上的队列，对其它服务器无效（因此这些队列不在主动目录中公布。）
  • 系统队列：包含日记队列（由系统生成）、死队列和事务型死信队列。死消息无法传送。

System.Messaging命名空间执行MSMQ的编程操作。这个命名空间有两个主要的对象：

  • Message：队列发送或读取的实际消息或数据。
  • MessageQueue：接收/发送消息的MSMQ消息队列。

二、消息队列类型

消息对列分为3类：
 
公共队列
 MachineName\QueueName
 能被别的机器所访问，如果你的多个项目中用到消息队列，那么你可以把队列定义为公共队列
 
专用队列
 MachineName\Private$\QueueName
 只针对于本机的程序才可以调用的队列，有些情况下为了安全起见定义为私有队列。
日志队列
 MachineName\QueueName\Journal$

然后编写以下代码：

消息的发送：

  class Program
    {
        static void Main(string[] args)
        {
            Message msg = null;
            MessageQueue mq = null;
            int count = 0;
            if (MessageQueue.Exists(@".\Private$\TechRepublic"))
            {
                MessageQueue.Delete(@".\Private$\TechRepublic");
            }
            while (true)
            {
                count++;
                
                try
                {
                    msg = new Message();
                    msg.Priority = MessagePriority.Normal;
                    if (!MessageQueue.Exists(@".\Private$\TechRepublic"))
                    {
                        mq = MessageQueue.Create(@".\Private$\TechRepublic");
                    }
                    else
                    {
                        mq = new MessageQueue(@".\Private$\TechRepublic");
                    }
                    msg.Label = "Test Message" + count;
                    Product p = new Product
                    {
                         Id = count,
                          Name = "name"+count,
                           CreateTime = DateTime.Now
                    };
                    msg.Body = p;
                    mq.Send(msg);
                    Console.WriteLine(p.Id);

                }
                catch (System.Messaging.MessageQueueException ex)
                {
                    Console.WriteLine("MSMQ Error: " + ex.ToString());
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Error: " + ex.ToString());
                }

            }


            #region 发送简单的类型
            //try
            //{
            //    mq = new MessageQueue(@".\Private$\TechRepublic");
            //    msg = mq.Receive(new TimeSpan(0, 0, 60));
            //    msg.Formatter = new XmlMessageFormatter(new String[] { "System.String,mscorlib" });
            //    //mq.Receive();
            //    Console.WriteLine(msg.Label.ToString() + " -- " + msg.Body.ToString());
            //}
            //catch (System.Messaging.MessageQueueException ex)
            //{
            //    Console.WriteLine("MSMQ Error: " + ex.ToString());
            //}
            //catch (Exception ex)
            //{
            //    Console.WriteLine("Error: " + ex.ToString());
            //}
            //finally
            //{
            //    mq.Close();
            //}
            #endregion
            Console.ReadKey();
        }
    }

消息的接收：

 class Program
    {
     

        static void Main(string[] args)
        {
               Message msg = null;
              MessageQueue mq = null;

            while (true)
            {
                Thread.Sleep(400);
                try
                {
                    if (MessageQueue.Exists(@".\Private$\TechRepublic"))
                    {
                        mq = new MessageQueue(@".\Private$\TechRepublic");
                        //#region 同步接收消息
                        //if (mq.GetAllMessages().Any())
                        //{
                        //msg = mq.Receive(new TimeSpan(0, 0, 60));
                        ////通过Receive方法接收消息同时永久性地从队列中删除消息；
                        ////通过Peek方法从队列中取出消息而不从队列中移除该消息。
                        //msg.Formatter = new XmlMessageFormatter(new String[] { "MyLib.Product,MyLib" });
                        //Product p = ((Product)msg.Body);
                        //Console.WriteLine(msg.Label.ToString() + " -- " + p.Id + "  ---" + p.CreateTime);
                        //}
                        //#endregion

                        #region 异步接收消息

                        //利用委托机制：
                        if (mq.GetAllMessages().Any())
                        {
                            mq.BeginReceive();
                            mq.ReceiveCompleted += mq_ReceiveCompleted;
                        }
                        #endregion



                    }

                }
                catch (System.Messaging.MessageQueueException ex)
                {
                    Console.WriteLine("MSMQ Error: " + ex.ToString());
                }
                catch (Exception ex)
                {
                    Console.WriteLine("Error: " + ex.ToString());
                }
            }

            Console.ReadKey();
        }

        static void mq_ReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
        {
            MessageQueue mq = (MessageQueue)sender;
            Message msg = mq.EndReceive(e.AsyncResult);
            msg.Formatter = new XmlMessageFormatter(new String[] { "MyLib.Product,MyLib" });
            Product p = ((Product)msg.Body);
            Console.WriteLine(msg.Label.ToString() + " -- " + p.Id + "  ---" + p.CreateTime);
            mq.BeginReceive();//如果希望继续异步接收消息，请在 ReceiveCompleted 事件处理程序中再次调用 BeginReceive 方法
        }
    }
namespace MyLib
{
    [Serializable]
    public class Product
    {
        public long Id { get; set; }
        public string Name { get; set; }
        public DateTime CreateTime { get; set; }
        public DateTime? UpdateTime { get; set; }
    }
}

 


联系
我们