深入浅出消息队列MSMQ

news2024/9/27 9:28:42

消息队列MSMQ，相信稍有开发经验的小伙伴都了解一些。开始讲解之前，我们先弄清楚一件事，为什么我们要使用MSMQ：

您可能认为您能够通过一个简单的数据库表(一个应用程序往其中写入数据，另一个应用程序从中读取数据)来应用消息队列。消息队列平台更为稳定，因为它们通常拥有自己的安全机制、事务支持及其它功能。传输消息的路由功能是它的一个关键应用。MSMQ提供各种消息队列平台。

总结一下MSMQ的优势是：高性能、跨平台和更好的安全机制、实物机制。同时，MSMQ这种消息中间件技术也是分布式开发的重要组成部分。

一、基本概念

MSMQ全称是Microsoft Message Queue——微软消息队列。它是一种异步传输模式，可以在不同的应用之间实现相互通信，相互通信的应用可以分布在同一台机器上，也可以分布于相连的网络空间中的任一位置。

二、工作原理

MSMQ的实现原理是：消息的发送者把自己想要发送的信息放入一个容器(Message)，然后把它保存到一个系统公用空间的消息队列(Message Queue)中，本地或异地的消息接收程序再从该队列中取出发给它的消息进行处理。

消息队列是一个公用存储空间，它可以存在于内存中或物理文件中，因此，消息以两种方式发送，即快递方式和可恢复模式。它们的区别是消息存储位置的不同，快递方式，为了消息的快速传递，所以把消息放置在内存中，而不放在物理磁盘上，以获得较高的处理能力；而可恢复模式在传送过程的每一步骤中，都把消息写入物理磁盘上，这样当保存消息队列的机器发生故障而重新启动后，可以把发送的消息恢复到故障发送之前的状态，以获得更好的消息恢复能力。消息队列可以放在发送方、接收方所在的机器上，也可以单独放置在另外一台机器上。另外，采用消息队列机制，发送方不必要担心接收方是否启动，是否发生故障等因素，只要消息成功发送出去，就可以认为处理完成，而实际上对方可能甚至未开机，或者实际消息传递到对方可能在第二天。MSMQ机制类似QQ消息传递机制。下图演示了MSMQ的实现原理。

MSMQ中主要有两个概念：

（1）一个是消息Message：Message是通信双方需要传递的消息，它可以是文本、图片、视频等。消息包含发送和接收者的标识，只有指定的用户才能取得消息。

（2）一个是队列Queue：用来保存消息的存储空间，MSMQ中主要包括以下几种队列类：

公共队列：在整个消息队列网络中复制，有可能由网络连接的所有站点访问。路径格式为：机器名称\队列名称
专用队列（或叫私有队列）：不在整个网络中发布，它们仅在所驻留的本地计算机上可用，专用队列只能由知道队列的完整路径名称或标签的应用程序访问。路径格式为：机器名称\Private$\队列名称
日志队列：包含确认在给定“消息队列中发送的消息回执消息”。路径格式为：机器名称\队列名称\Journal$
响应队列：包含目标应用程序接收到消息时返回给发送应用程序的响应消息，包括机器日志队列、机器死信队列和机器事务死信队列。其中，机器信道死信队列对应的格式为：机器名称\XactDeadletter$。机器死信队列对应的格式为：机器名称\Deadletter$；机器日志队列对应的格式为：机器名称\Journal$；

三、MSMQ的安装和配置

打开Control Panel-“Add/Remove Programs” – “Add/Remove Windows Components”步骤安装MSMQ。

MSMQ可以安装为工作组模式或域模式。如果安装程序没有找到一台运行提供目录服务的消息队列的服务器，则只可以安装为工作组模式，此计算机上的“消息队列”只支持创建专用队列和创建与其他运行“消息队列”的计算机的直接连接。

配置MSMQ：打开Computer Management – Message Queuing，在Private Queues下创建MSMQDemo队列

四、消息发送和消费实例

1、消息发送

首先从发送数据开始，在发送数据时首先要创建我们的MQ，然后根据MQ的地址创建相应的队列，调用队列的send方法将数据信息发送到队列中，如下代码：

using System;  
using System.Collections.Generic;  
using System.Linq;  
using System.Linq.Expressions;  
using System.Messaging;  
using System.Runtime.CompilerServices;  
using System.Text;  
using System.Threading;  
using System.Xml;  
using System.Xml.Serialization;  
using System.Runtime.Serialization.Formatters.Binary;  
  
namespace MsMQTest  
{  
    class Program  
    {  
        static void Main(string[] args)  
        {  
            //declare the MQ Path  
            string ekQ= ".\\Private$\\EKTestQueue";  
  
            //create the MQ if the MQ is not exist  
            if (!MessageQueue.Exists(ekQ))  
                MessageQueue.Create(ekQ);  
  
            //create a new queue  
            var queue = new MessageQueue(ekQ);  
  
            for (int i = 0; i < 2; i++)  
            {  
                //create the model that want to send  
                Test test=new Test();  
                test.Name = "fdsfd";  
                test.Sex = "cvx";  
                //serialize the model  
                string str = Program.xmlSerial(test);  
                //send the model data to queue  
                queue.Send("Test" + str);  
                Console.WriteLine("Message sent {0} \n--------------", "Test" +str);  
            }  
  
            Console.Read();  
  
            // MessageQueue.Delete(ekQ);  
        }  
  
        public static string xmlSerial<T>(T serializeClass)  
        {  
            string xmlString = string.Empty;  
            XmlWriterSettings settings = new XmlWriterSettings();  
            XmlSerializer serializer = new XmlSerializer(typeof(T));  
            StringBuilder xmlStringBuilder = new StringBuilder();  
            using (XmlWriter writer = XmlWriter.Create(xmlStringBuilder))  
            {  
                serializer.Serialize(writer, serializeClass);  
                xmlString = xmlStringBuilder.ToString();  
            }  
  
            return xmlString;  
        }  
    }  
  
    public class Test  
    {  
        public string Name { get; set; }  
        public string Sex { get; set; }  
    }  
}

运行结果：

MSMQ消息截图：

2、消息接收

运行上面的代码后会把消息发送到相应的消息队列中，这样在消息的发送方和调用方之间就构建了一个相互松耦合的桥梁，它就是消息队列，接下来演示如何接收消息队列。

using System;  
using System.Collections.Generic;  
using System.Linq;  
using System.Messaging;  
using System.Text;  
using System.Threading;  
namespace MsqueueReaderTest  
{  
    class Program  
    {  
        static void Main(string[] args)  
        {  
            string ekQ = ".\\Private$\\EKTestQueue";  
              
            using (var queue = new MessageQueue(ekQ))  
            {  
                queue.Formatter = new XmlMessageFormatter(new Type[] { typeof(String) });  
                var exist = false;  
                while (!MessageQueue.Exists(ekQ))  
                {  
                    Console.WriteLine("No existing queue");  
                          
                }  
                exist = true;  
                while (exist)  
                {  
                    var m = queue.Receive();  
                    Console.WriteLine("Message Received {0} \n--------------",(string)m.Body);  
                    // Thread.Sleep(500);  
                }  
            }  
              
        }  
    }  
}

运行结果如下：