MSMQ消息队列

消息队列（MSMQ）技术使得运行于不同时间的应用程序能够在各种各样的网络和可能暂时脱机的系统之间进行通信。

应用程序将消息发送到队列，并从队列中读取消息。

消息即是信息的载体。为了让消息发送者和消息接收者都能够明白消息所承载的信息（消息发送者需要知道如何构造消息；消息接收者需要知道如何解析消息）

下图演示了消息队列如何保存由多个发送应用程序生成的消息，并被多个接收应用程序读取。

消息队列的安装可以

1、安装消息队列
   开始—》控制面板—》管理工具—》服务器管理器—》功能—》添加功能—》依次展开MSM、MSMQ服务—》确定。
2、管理消息队列
   计算机—》右键—》管理—》功能—》消息队列。

参考 点击

我理解的MSMQ

      MSMQ可以被看成一个数据储存装置，就如同数据库，只不过数据存储的是一条一条的记录，而MSMQ存储的是一个一个的消息(messsge)。

Message可以被理解为一种数据容器，我们在稍后会讲到。MSMQ一个重要的应用场景就是离线信息交互，例如，我们在给朋友发送邮件，而此时朋友并未登入邮箱，

这个时候我们的邮件就可以发到邮件服务器的MSMQ队列中，当朋友登入邮箱的时候，系统在从服务器的MSMQ队列中取出U件。当然MSMQ的用途远不止这些，

例如，充当数据缓存，实现异步操作等等，这里就不在一一举例了。

创建、删除和管理队列
   要开发MSMQ程序就必须学习一个很重要的类（MessageQueue），该类位于名称空间System.Messageing下。
常用方法：
   --Create() 　　　　　　  　　 方法:创建使用指定路径的新消息队列。
   --Delete() 方法：　　　  　　 删除现有的消息队列。
   --Existe() 方法：　　　　　　 查看指定消息队列是否存在。
   --GetAllMessages() 方法 :　   得到队列中的所有消息。
   --GetPublicQueues() 方法：   在“消息队列”网络中定位消息队列。
   --Peek()/BeginPeek() 方法：  查看某个特定队列中的消息队列，但不从该队列中移出消息。
   --Receive()/BeginReceive()    方法：检索指定消息队列中最前面的消息并将其从该队列中移除。
   --Send() 　　　　　　　　　 方法：发送消息到指定的消息队列。
   --Purge() 　　　　　　　　   方法：清空指定队列的消息。
常用属性：
   Priority : 设置消息优先级，MessagePriority 枚举里全部进行了封装，MessagePriority.High()；
   AboveNormal：hight 与Normal 消息优先级之间；
   High：高级消息优先级；
   Highest：最高消息优先级；
   Low：低消息优先级；
   Lowest：最低消息优先级；
   Normal：普通消息优先级；
   VeryHigh：Highest 和High 消息优先级之间；
   VeryLow：Low 和Lowest 消息优先级之间；
发送和序列化消息
   MSMQ消息队列中定义的消息由一个主体(body)和若干属性构成。消息的主体可以由文本、二进制构成，根据需要还可以被加密。
   在MSMQ中消息的大小不能够超过4MB。发送消息是通过Send方法来完成的，需要一个Message参数。
1、发送消息：
   步骤：连接队列-->指定消息格式-->提供要发送的数据（主体）-->调用Send()方法将消息发送出去。详细见后面的示例程序。
2、序列化消息：
   消息序列化可以通过.NET Framework附带的三个预定义格式化程序来完成：
   --  XMLMessageFormatter对象----MessageQueue组件的默认格式化程序设置。
   --  BinaryMessageFormatter对象；
   --  ActiveXMessageFormatter对象； 
   由于后两者格式化后的消息通常不能为人阅读，所以我们经常用到的是XMLMessageFormatter对象。该对象构造方法有三种重载：
   1、public XmlMessageFormatter();
   2、public XmlMessageFormatter(string[] targetTypeNames);
   3、public XmlMessageFormatter(Type[] targetTypes);
   如我们后面的示例程序中用到的序列化语句：
   //序列化为字符串
   XmlMessageFormatter formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
读取和接收消息
1、读取消息：
   也就是从指定队列中获取消息。
2、接收消息有两种方式：
   --> 通过 Receive() 方法。获取移除消息。
   --> 通过 Peek() 方法。 获取但不移除消息。

　　路径名和格式名可以用于标识队列，要查找队列，必须区分公共队列和私有队列。

　　公共队列在Active Directory中发布，对于这些队列，无需知道他们所在的系统，私有队列只有在已知队列所在的系统名时才能找到。
　　在Active Directory域中收索队列的标签，类别或格式名，就可以找到公共队列，还可以获得机器上的所有队列，

　　MessageQueue类的静态方法 GetPublicQueuesByLabel()、GetPublicQueuesByCategory() 和 GetPublicQueuesByMachine() 

　　可以收索队列，GetPublicQueues() 方法返回包含域中所有公共队列的数组。

如：

static void Main(string[] args)
{
 foreach(MessageQueue queue in MessageQueue.GetPublicQueues())
 {
   Console.WriteLine(queue.Path);
 }
}

消息队列分为以下几种，每种队列的路径表示形式如下：

这里的  MachineName  可以用 “."代替,代表当前计算机

需要先引用System.Messaging.dll

MSMQ 支持两种类型的队列，

1、事务性队列(transactional queue)会将消息持久(persiste)存储到磁盘中，即便服务器当机(shutdown)、重启(reboot)或崩溃(crash)，

消息依然 可以在系统恢复后被读取。同时，消息发布、获取和删除都在环境事务范围内，从而确保消息的可靠性。

我们还可以使用 TransactionScope 将环境事务传递给队列，否则队列会自动创建一个内部事务。

2、非事务性队列(nontransactional volatile queues)只是将消息存在内存，不会使用磁盘进行持久存储，且不会使用事务来保护对消息的操作。

一但服务器发生问题，或者调用方出现异常，消息都会丢 失。

// 创建事务性队列

MessageQueue.Create(@"./private$/myqueue", true);

// 创建非事务性队列

MessageQueue.Create(@"./private$/myqueue");

3、消息队列的优先级

　　在MSMQ中消息在队列里传输是分有优先级的，　　

　　优先级一共有七种，MessagePriority枚举里全部进行了封装。因这里只作程序演示就不一一列举出，仅用了Highest和Normal两种类型，

　　关于消息队列上进行消息传输的七种优先级大家可以参考我下面提供的 MessagePriority 枚举源代码定义。

　　那么在发送消息的时候怎么指定消息的优先级呢？在Message对象里封装有一个属性Priority，

　　接受一个枚举MessagePriority类型的值来设置消息传输的优先级。如下：

Message message = new Message();
message.Priority = MessagePriority.Lowest;  //定义消息的优先级

 　　消息的优先级不能用于事务性消息发送。因为事务性消息是按照顺序发生和接收消息的。

4、消息队列超时设置和最大长度设置

一、非事务性消息发生接收

发送：

    class Program
    {
        static void Main(string[] args)
        {
            string msg = Console.ReadLine();
            //发生消息
            new SendQueueMessage().SendMessageQueue(msg);
            Console.ReadLine();
        }
   }    

    /// <summary>
    /// 消息队列发生消息
    /// </summary>
    public class SendQueueMessage
    {
        /// <summary>
        /// 消息队列
        /// </summary>
        private MessageQueue messageQueue;

        public SendQueueMessage()
        {
            //路径
            string path = ".\private$\temp";
            if (MessageQueue.Exists(path))
            {
                //如果存在指定路径的消息队列，则获取
                messageQueue = new MessageQueue(path);
            }
            else
            {
                //不存在，则创建新的
                messageQueue = MessageQueue.Create(path);
            }
        }

        /// <summary>
        /// 发生消息
        /// </summary>
        /// <param name="msg"></param>
        public void SendMessageQueue(string msg)
        {
            Message message = new Message();
            message.Body = msg;
            //指定格式
            message.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
            messageQueue.Send(message);
        }
    }

接收：

    class Program
    {
        static void Main(string[] args)
        {
            //接收
            new ReciveQueue().ReciveMessage();
            Console.ReadLine();
        }
    }

    /// <summary>
    /// 消息队列接收
    /// </summary>
    public class ReciveQueue
    {
        /// <summary>
        /// 消息队列
        /// </summary>
        private MessageQueue messageQueue;

        public ReciveQueue()
        {
            //路径
            string path = ".\private$\temp";
            if (MessageQueue.Exists(path))
            {
                //如果存在指定路径的消息队列，则获取
                messageQueue = new MessageQueue(path);
            }
            else
            {
                //不存在，则创建新的
                messageQueue = MessageQueue.Create(path);
            }
        }

        public void ReciveMessage()
        {
            while (true)
            {
                //Message.Receive()是同步进行的,如果队列中没有消息，会阻塞当前线程
                Message message = messageQueue.Receive();
                message.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
                string msg = message.Body.ToString();
                Console.WriteLine(msg);
            }
        }
    }

注意：messageQueue.Receive()  如果没有接收到消息，则会阻塞当前被调用线程，直到接收到消息为止。

 这样就在数据转发服务器端创建了一个名为  temp 的消息队列；从客户端要发送的消息就保存在这个队列里，

你可以通过计算机管理->服务和应用下的消息队列中看到你创建的 temp 队列，private$关键字是说明队列为专用队列，

前面的“.”代表创建的队列目录是本机，这个队列一旦创建成功，就是系统的事了，接下来要做的就是你怎么去把消息写进这个队列，或者读取队列的值 

这里要特别注意，不要将queuepath路径字符串写成 

        /// <summary>
        /// 获取专用消息队列
        /// </summary>
        /// <returns></returns>
        public static MessageQueue GetMessageQueue()
        {
            string publicMQPath = "FormatName:Direct=TCP:192.168.0.110\private$\DESKTOP-2AEJ0GU";
            //如果存在，获取消息队列
            if (MessageQueue.Exists(publicMQPath))
            {
                queue = new MessageQueue(publicMQPath);
            }
            else
            {
                //不存在，创建一个新的消息队列
                queue = MessageQueue.Create(publicMQPath);
            }
            return queue;
        }

这样写的话是用于远程计算机对这个队列进行访问的，

因为MessageQueue的Create() 和 Exisit() 方法是没办法去识别上述FormatName格式的，

还有要确保 Create() 函数要被执行了之后再用 MessageQueue 实例去引用；这样服务器端队列的创建就完成了；

二、事务性消息队列

　　事务我想大家对这个词应该都不会陌生，在操作数据库的时候经常都会用到事务，确保操作成功，要么全部完成（成功） 
，要么全部不完成（失败）。在MSMQ中利用事务性处理，可以确保事务中的消息按照顺序传送，只传送一次，并且从目的队列成 
功地被检索。
　　那么，在MSMQ上使用事务性处理怎么实现呢？可以通过创建 MessageQueueTransation 类的实例并将其关联到 MessageQueue 
组件的实例来执行，执行事务的 Begin 方法，并将其实例传递到收发方法。然后，调用 Commit 以将事务的更改保存到目的队列。
创建事务性消息和普通的消息有一点小小的区别，大家可从下图上体会到：

 发送：

    /// <summary>
    /// 事务性消息发送
    /// </summary>
    public class MQTranSendTest
    {
        /// <summary>
        /// 执行发送
        /// </summary>
        public void Execute()
        {
            Console.WriteLine("请输入发送消息:");
            string flag = "N";
            while (flag != "Y")
            {
                string msg = Console.ReadLine();
                SendMessage(msg);
                Console.WriteLine("按回车键继续，或者输入 Y 退出程序!");
                flag = Console.ReadLine();
            }
        }

        /// <summary>
        /// 格式化发生内容，执行发送
        /// </summary>
        /// <param name="msg">发送的消息</param>
        private static void SendMessage(string msg)
        {
            Message message = new Message();
            MessageQueue queue = GetMessageQueue();
            message.Body = msg;
            //指定格式
            queue.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
            MessageQueueTransaction queueTransaction = new MessageQueueTransaction();
            //启动事务
            queueTransaction.Begin();
            queue.Send(message, queueTransaction);
            //提交事务
            queueTransaction.Commit();
        }

        /// <summary>
        /// 根据路径获取消息队列
        /// </summary>
        /// <returns></returns>
        private static MessageQueue GetMessageQueue()
        {
            string path = ".\private$\TranTemp";
            if (MessageQueue.Exists(path))
            {
                //如果存在指定路径的消息队列，则获取
                return new MessageQueue(path);
            }
            else
            {
                //不存在，则创建新的
                return MessageQueue.Create(path,true);
            }
        }
    }

接收：

 /// <summary>
    /// 事务性接收消息
    /// </summary>
    public class MQTranReciveTest
    {
        /// <summary>
        /// 消息队列
        /// </summary>
        private MessageQueue messageQueue;

        public MQTranReciveTest()
        {
            //路径
            string path = ".\private$\TranTemp";
            if (MessageQueue.Exists(path))
            {
                //如果存在指定路径的消息队列，则获取
                messageQueue = new MessageQueue(path);
            }
            else
            {
                //不存在，则创建新的
                messageQueue = MessageQueue.Create(path);
            }
        }

        public void ReciveMessage()
        {
            string msg = string.Empty;
            while (true)
            {
                if (messageQueue.Transactional)
                {
                    MessageQueueTransaction queueTransaction = new MessageQueueTransaction();
                    //启动事务
                    queueTransaction.Begin();
                    //Message.Receive()是同步进行的,如果队列中没有消息，会阻塞当前线程
                    Message message = messageQueue.Receive(queueTransaction);
                    message.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
                    msg = message.Body.ToString();
                    queueTransaction.Commit();
                    Console.WriteLine(msg);
                    msg = string.Empty;
                }
            }
        }
    }

二、死信队列

 死信队列：由于某种原因无法传递的消息都放置在死信队列上（如：超时，超出消息队列最大长度等）

 超时、消息队列长度设置代码：

　　　　　/// <summary>
        /// 格式化发生内容，执行发送
        /// </summary>
        /// <param name="msg">发送的消息</param>
        public static void SendMessage(string msg)
        {
            Message message = new Message(msg);
            MessageQueue queue = GetMainQueue();
            message.Body = msg;
            //指定格式
            queue.Formatter = new XmlMessageFormatter(new Type[] { typeof(string) });
            //设置日志队列最大个数
            //queue.MaximumJournalSize = 1;
            //设置消息队列最大存放的个数
            //queue.MaximumQueueSize = 1;
            //发送消息的时间开始算起，消息到达目标队列的时间
            //message.TimeToReachQueue = new TimeSpan(0, 0, 1);
            //从发送消息到接收消息的总时间
            //message.TimeToBeReceived = new TimeSpan(0, 0, 30);
            //设置为true，发送失败后进入死信队列
            //message.UseDeadLetterQueue = true;
            queue.Send(message);
        }

如何访问远程的私有队列?
如何修改消息队列的最大存储限制?
如何修改消息队列的默认存储位置?
如何保证计算机重启之后队列中的消息还在?

文档：https://files.cnblogs.com/files/wwj1992/MSMQ_FAQ.rar

　　  https://files.cnblogs.com/files/wwj1992/MSMQ_Doc.rar

参考：https://www.cnblogs.com/Leo_wl/p/4021039.html

Demo 下载