MSMQ(消息队列)学习笔记

MSMQ Message Queue 微软消息队列 是在多个不同的应用之间实现相互通信的一种异步 传输模式 相互通信的应用可以分布于同一台机器上 也可以分布于相连的网络空间中的 任一位置 它的实现原理是 消息的发送者把自己想要发送的信息放入一个容器中 我们称之为 Message 然后把它保存至一个系统公用空间的消息队列 Message Queue 中 本地或 者是异地的消息接收程序再从该队列中取出发给它的消息进行处理 一 特点一 特点 消息处理和消息为基于服务器的应用程序组件之间的进程间通信提供了强大灵活的机 制 同组件间的直接调用相比 它们具有若干优点 其中包括 稳定性 组件失败对消息的影响程度远远小于组件间的直接调用 因为消息存储 在队列中并一直留在那里 直到被适当地处理 消息处理同事务处理相似 因为消 息处理是有保证的 消息优先级 更紧急或更重要的消息可在相对不重要的消息之前接收 因此可以 为关键的应用程序保证足够的响应时间 脱机能力 发送消息时 它们可被发送到临时队列中并一直留在那里 直到被成 功地传递 当因任何原因对所需队列的访问不可用时 用户可以继续执行操作 同 时 其他操作可以继续进行 如同消息已经得到了处理一样 这是因为网络连接恢 复时消息传递是有保证的 事务性消息处理 将多个相关消息耦合为单个事务 确保消息按顺序传递 只传 递一次并且可以从它们的目标队列中被成功地检索 如果出现任何错误 将取消整 个事务 安全性 MessageQueue 组件基于的消息队列技术使用 Windows 安全来保 护访问控制 提供审核 并对组件发送和接收的消息进行加密和验证 故障恢复能力 消息可以以两种方式发送 即快递方式 express 和可恢复模式 recoverable 它们的区别在于 快递方式为了消息的快速传递 把消息放置 于内存中 而不放于物理磁盘上 以获取较高的处理能力 可恢复模式在传送过程 的每一步骤中 都把消息写入物理磁盘中 以得到较好的故障恢复能力 MSMQ 与 XML Web Services 和 Net Remoting 一样 是一种分布式开发技术 但是在使用 XML Web Services 或 Net Remoting 组件时 Client 端需要和 Server 端 实时交换信息 Server 需要保持联机 MSMQ 则可以在 Server 离线的情况下工作 将 Message 临时保存在 Client 端的消息队列中 以后联机时再发送到 Server 端处理 虽然 XML Web Services 和 Net Remoting 都提供了 OneWay 属性来处理异步调 用 用来解决 Server 端长方法调用 长时间阻碍 Client 端 但是不能解决大量 Client 负载的问题 此时 Server 接受的请求快于处理请求 二 名词解释二 名词解释 1 消息消息 消息是由通信的双方所需要传递的信息 它可以是各式各样的媒体 如文本 声音 图象等等 消息最终的理解方式 为消息传递的双方事先商定 这样做的好处是 一是相当于对 数据进行了简单的加密 二则采用自己定义的格式可以节省通信的传递量 消息可以含有发送和接收者的标识 这样只有指定的用户才能看到只传递给他的信息 和返回是否操作成功的回执 消息也可以含有时间戳 以便于接收方对某些与时间相关的 应用进行处理 消息还可以含有到期时间 它表明如果在指定时间内消息还未到达则作废 这主要应用与时间性关联较为紧密的应用 2 队列队列 队列是发送和接收消息的公用存储空间 它可以存在于内存中或者是物理文件中 有两种主要的队列类型 由您或网络中的其他用户创建的队列和系统队列 1 用户队列 公共队列 在整个 消息队列 网络中复制 并且有可能由网络连接的所有站点访 问 专用队列 不在整个网络中发布 相反 它们仅在所驻留的本地计算机上可用 专 用队列只能由知道队列的完整路径名或标签的应用程序访问 管理队列 包含确认在给定 消息队列 网络中发送的消息回执的消息 指定希望 MessageQueue 组件使用的管理队列 如果有的话 响应队列 包含目标应用程序接收到消息时返回给发送应用程序的响应消息 指定 希望 MessageQueue 组件使用的响应队列 如果有的话 2 系统队列 日记队列 可选地存储发送消息的副本和从队列中移除的消息副本 每个 消息队 列 客户端上的单个日记队列存储从该计算机发送的消息副本 在服务器上为每个队列创 建了一个单独的日记队列 此日记跟踪从该队列中移除的消息 死信队列 存储无法传递或已过期的消息的副本 如果过期或无法传递的消息是事 务性消息 则被存储在一种特殊的死信队列中 称为 事务性死信队列 死信存储在过 期消息所在的计算机上 有关超时期限和过期消息的更多信息 请参见默认消息属性 报告队列 包含指示消息到达目标所经过的路由的消息 还可以包含测试消息 每 台计算机上只能有一个报告队列 专用系统队列 是一系列存储系统执行消息处理操作所需的管理和通知消息的专用 队列 在应用程序中进行的大多数工作都涉及访问公共队列及其消息 但是 根据应用程序 的日记记录 确认和其他特殊处理需要 在日常操作中很可能要使用几种不同的系统队列 三 安装消息队列三 安装消息队列 使用 MSMQ 进行软件开发需要安装 MSMQ 安装完后就该进入实际的开发阶段 具 体的安装过程就是在控制面板里 添加 删除程序 下 添加 删除 Windows 组件 完成添 加就 OK 安装完成后就可以通过交互界添加新的消息队列 详细如下图 四 创建 删除和管理队列四 创建 删除和管理队列 API 要开发 MSMQ 程序就必须学习一个很重要的类 MessageQueue 该类位于名称空 间 System Messageing 下 其中有几个常用的方法必须掌握 Create 方法 创建使用指定路径的新消息队列 Delete 方法 删除现有的消息队列 Existe 方法 查看指定消息队列是否存在 GetAllMessages 方法 得到队列中的所有消息 GetPublicQueues 方法 在 消息队列 网络中定位消息队列 Peek BeginPeek 方法 查看某个特定队列中的消息队列 但不从该队列中移出消息 Receive BeginReceive 方法 检索指定消息队列中最前面的消息并将其从该队列中移除 Send 方法 发送消息到指定的消息队列 Purge 方法 清空指定队列的消息 五 消息处理流程五 消息处理流程 MSMQ 消息队列中定义的消息由一个主体 body 和若干属性构成 消息的主体可以由 文本 二进制构成 根据需要还可以被加密 在 MSMQ 中消息的大小不能够超过 4MB 1 1 创建消息创建消息 NET 框架里的 MessageQueue 类下有一静态方法 Create 用来完成消息队列的创建 path 要创建的队列的路径 transactional 事务性队列 为 true 非事务性队列 则为 false public static MessageQueue Create string path public static MessageQueue Create string path bool transactional 事务性消息 可以确保事务中的消息按照顺序传送 只传送一次 并且从目的队列成 功地被检索 MessageQueueAccessControlEntry 为受信者 用户 组或计算机 指定执行消息队列 任务的访问权限 附加 Everyone 访问权限 MessageQueueAccessControlEntry accessControl new MessageQueueAccessControlEntry new Trustee Everyone MessageQueueAccessRights FullControl myQueue SetPermissions accessControl eg 通过 Create 方法创建使用指定路径的新消息队列 通过Create方法创建使用指定路径的新消息队列 public static MessageQueue Createqueue string queuePath try if MessageQueue Exists queuePath MessageQueue myTranMessage MessageQueue Create private myQueue 创建事务性的专用消息队列 MessageQueue myTranMessage MessageQueue Create private myQueueTrans true else catch MessageQueueException e 2 2 发送消息发送消息 发送消息是通过 Send 方法来完成的 需要一个 Message 参数 步骤 a 连接队列 b 指定消息发送的优先级 c 指定消息格式 d 提供要发送的数据 主体 e 调用 Send 方法将消息发送出去 简单消息类型就是常用的数据类型 例如整型 字符串等数据 复杂消息的数据类型 通常对应于系统中的复杂数据类型 例如结构 对象等等 简单消息的发送示例如下 mq send 1000 发送整型数据 mq send this is a test message 发送字符串 复杂消息的发送和简单消息的发送大同小异 只是发送时 通常不是直接给出发送的 消息内容 而是代表发送消息内容的变量 下面的代码中发送的消息由复杂数据类型变量表示 Customer为自定义的一个类 Customer customer new Customer customer FirstName copernicus customer LastName nicolaus mq send customer 3 3 设置发送优先级设置发送优先级 在 MSMQ 中消息在队列里传输是分有优先级的 优先级一共有七种 枚举 MessagePriority 里进行了封装 public enum MessagePriority Lowest 0 VeryLow 1 Low 2 Normal 3 AboveNormal 4 High 5 VeryHigh 6 Highest 7 设置优先级 message Priority MessagePriority Highest 最高消息优先级 4 4 格式化消息格式化消息 消息序列化可以通过 NET Framework 附带的三个预定义格式化程序来完成 XMLMessageFormatter 对象 MessageQueue 组件的默认格式化程序设置 BinaryMessageFormatter 对象 ActiveXMessageFormatter 对象 由于后两者格式化后的消息通常不能为人阅读 所以我们经常用到的是 XMLMessageFormatter 对象 该对象构造方法有三种重载 public XmlMessageFormatter public XmlMessageFormatter string targetTypeNames public XmlMessageFormatter Type targetTypes eg 连接消息队列并发送消息到队列 public static void SendMessage try 连接到本地的队列 MessageQueue myQueue new MessageQueue private myQueue Message myMessage new Message myMessage Body 消息内容 设置最高消息优先级 message Priority MessagePriority Highest 序列化为字符串 myMessage Formatter new XmlMessageFormatter new Type typeof string 发送消息到队列中 myQueue Send myMessage 事务性消息需加上下面几句 MessageQueueTransaction myTransaction new MessageQueueTransaction 启动事务 myTransaction Begin myQueue Send myMessage myTransaction 加了事务 提交事务 myTransaction Commit catch ArgumentException e 5 5 接收消息接收消息 Receive 方法接收消息同时永久性地从队列中删除消息 Peek 方法从队列中取出消息而不从队列中移除该消息 如果知道消息的标识符 ID 还可以通过 ReceiveById 方法和 PeekById 方法完 成相应的操作 连接消息队列并从队列中接收消息 public static void ReceiveMessage 连接到本地队列 MessageQueue myQueue new MessageQueue private myQueue 注 由于消息的优先级是枚举类型 在直接messages index Priority ToString 这种方式来获取优先 级转化到字符串的时候 他需要一个过滤器 Filter 否则会抛出一个InvalidCastExceptionle类型的异常 异 常信息 接收消息时未检索到属性 Priority 请确保正确设置了 PropertyFilter 要解决这问题只需要把 消息对象的MessageReadPropertyFilter 过滤器 的Priority设置为true myQueue MessageReadPropertyFilter Priority true 接收端必须反序列化 myQueue Formatter new XmlMessageFormatter new Type typeof string try if myQueue Transactional MessageQueueTransaction myTransaction new MessageQueueTransaction 启动事务 myTransaction Begin 从队列中接收消息 Message myMessage myQueue Receive myTransaction string context string myMessage Body 获取消息的内容 string priority messages index Priority ToString myTransaction Commit else 从队列中接收消息 Message myMessage myQueue Receive string context string myMessage Body 获取消息的内容 string priority messages index Priority ToString catch MessageQueueException e catch InvalidCastException e 连接队列并获取队列的全部消息 public static void GetAllMessage 连接到本地队列 MessageQueue myQueue new MessageQueue private myQueue Message message myQueue GetAllMessages XmlMessageFormatter formatter new XmlMessageFormatter new Type typeof string for int i 0 i message Length i message i Formatter formatter Console WriteLine message i Body ToString 6 6 异步消息处理异步消息处理 在 MSMQ 中对消息的操作分有同步化操作和异步化操作两种 那两者到底有何区别 呢 简单的说同步化操作就是一项操作没有完成前它将堵塞整个进程直到操作完成 下一 项操作才会执行 所谓异步化操作则相反 不会堵塞启动操作的调用线程 如果你想在检 索消息但不想堵塞其他程序的执行 则可使用异步消息处理 在 MSMQ 中异步接收消息使用 BeginReceive 方法和 EndReceive 方法来标记操作的开 始和结束 异步查看消息则使用 BeginPeek 和 EndPeek 两个方法来标记异步读取的开始和 结束 同时 异步接收和查看消息还可以指定超时时间 发送消息需加上事务处理 同上 异步接收消息 利用委托机制 myQueue ReceiveCompleted newReceiveCompletedEventHandler MyReceiveCompleted 如下 通知一个或多个正在等待的线程已发生事件 static ManualResetEvent signal new ManualResetEvent