Framework——【MessageQueue】消息队列

定义

队列是 Apache RocketMQ 中消息存储和传输的实际容器，也是 Apache RocketMQ 消息的最小存储单元。 Apache RocketMQ 的所有主题都是由多个队列组成，以此实现队列数量的水平拆分和队列内部的流式存储。

队列的主要作用如下：

• 存储顺序性 队列天然具备顺序性，即消息按照进入队列的顺序写入存储，同一队列间的消息天然存在顺序关系，队列头部为最早写入的消息，队列尾部为最新写入的消息。消息在队列中的位置和消息之间的顺序通过位点（Offset）进行标记管理。
• 流式操作语义 Apache RocketMQ 基于队列的存储模型可确保消息从任意位点读取任意数量的消息，以此实现类似聚合读取、回溯读取等特性，这些特性是RabbitMQ、ActiveMQ等非队列存储模型不具备的。

模型关系

在整个 Apache RocketMQ 的领域模型中，队列所处的流程和位置如下：

Apache RocketMQ 默认提供消息可靠存储机制，所有发送成功的消息都被持久化存储到队列中，配合生产者和消费者客户端的调用可实现至少投递一次的可靠性语义。

Apache RocketMQ 队列模型和Kafka的分区（Partition）模型类似。在 Apache RocketMQ 消息收发模型中，队列属于主题的一部分，虽然所有的消息资源以主题粒度管理，但实际的操作实现是面向队列。例如，生产者指定某个主题，向主题内发送消息，但实际消息发送到该主题下的某个队列中。

Apache RocketMQ 中通过修改队列数量，以此实现横向的水平扩容和缩容。

内部属性

读写权限

• 定义：当前队列是否可以读写数据。
• 取值：由服务端定义，枚举值如下
• 6：读写状态，当前队列允许读取消息和写入消息。
• 4：只读状态，当前队列只允许读取消息，不允许写入消息。
• 2：只写状态，当前队列只允许写入消息，不允许读取消息。
• 0：不可读写状态，当前队列不允许读取消息和写入消息。
• 约束：队列的读写权限属于运维侧操作，不建议频繁修改。

行为约束

每个主题下会由一到多个队列来存储消息，每个主题对应的队列数与消息类型以及实例所处地域（Region）相关，队列数暂不支持修改。

版本兼容性

队列的名称属性在 Apache RocketMQ 服务端的不同版本中有如下差异：

• 服务端3.x/4.x版本：队列名称由{主题名称}+{BrokerID}+{QueueID}三元组组成，和物理节点绑定。
• 服务端5.x版本：队列名称为一个集群分配的全局唯一的字符串组成，和物理节点解耦。

因此，在开发过程中，建议不要对队列名称做任何假设和绑定。如果您在代码中自定义拼接队列名称并和其他操作进行绑定，一旦服务端版本升级，可能会出现队列名称无法解析的兼容性问题。

使用建议

按照实际业务消耗设置队列数

Apache RocketMQ 的队列数可在创建主题或变更主题时设置修改，队列数量的设置应遵循少用够用原则，避免随意增加队列数量。

主题内队列数过多可能对导致如下问题：

• 集群元数据膨胀 Apache RocketMQ 会以队列粒度采集指标和监控数据，队列过多容易造成管控元数据膨胀。
• 客户端压力过大 Apache RocketMQ 的消息读写都是针对队列进行操作，队列过多容易产生空轮询请求，增加系统负荷。

常见队列增加场景

• 需要增加队列实现物理节点负载均衡 Apache RocketMQ 每个主题的多个队列可以分布在不同的服务节点上，在集群水平扩容增加节点后，为了保证集群流量的负载均衡，建议在新的服务节点上新增队列，或将旧的队列迁移到新的服务节点上。
• 需要增加队列实现顺序消息性能扩展 在 Apache RocketMQ 服务端4.x版本中，顺序消息的顺序性在队列内生效的，因此顺序消息的并发度会在一定程度上受队列数量的影响，因此建议仅在系统性能瓶颈时再增加队列。

消息队列相关概念

1. 生产者（Producer）： 负责产生消息；
2. 消费者（Consumer）： 负责消费消息；
3. 消息（Message）： 在应用间传送的数据。消息可以非常简单，比如只包含文本字符串，也可以更复杂，可能包含嵌入对象；
4. 消息队列（Message Queue）： 一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。
5. 消息代理（Message Broker）： 负责存储/转发消息，转发分为推和拉两种。

• 拉是指Consumer主动从Message Broker获取消息；
• 推是指Message Broker主动将Consumer感兴趣的消息推送给Consumer。

消息队列的消费场景

消息至多被消费一次

该场景是最容易满足的，特点是整个消息队列吞吐量大，实现简单。适合能容忍丢消息，消息重复消费的任务。

a）Producer发送消息到Message Broker阶段：

Producer发消息给Message Broker，不要求Message Broker对接收到的消息响应确认，Producer也不用关心Message Broker是否收到消息了。

b）Message Broker存储/转发阶段：

对Message Broker的存储不要求持久性，转发消息时也不用关心Consumer是否真的收到了。

c）Consumer消费阶段：

Consumer从Message Broker中获取到消息后，可以从Message Broker删除消息，或Message Broker在消息被Consumer拿去消费时删除消息，不用关心Consumer最后对消息的处理结果。

消息至少被消费一次

适合不能容忍丢消息，但允许重复消费的任务。

a）Producer发送消息到Message Broker阶段：

Producer发消息给Message Broker，Message Broker必须响应对消息的确认。

b）Message Broker存储/转发阶段：

Message Broker必须提供持久性保障，转发消息时，Message Broker需要Consumer通知删除消息，才能将消息删除。

c）Consumer消费阶段：

Consumer从Message Broker中获取到消息，必须在消费完成后，Message Broker上的消息才能被删除。

消息仅被消费一次

适合对消息消费情况要求非常高的任务，实现较为复杂，这里的“仅被消费一次”包含如下两种场景：

1）Message Broker上存储的消息被Consumer仅消费一次，场景要求：

a）Producer发送消息到Message Broker阶段：

Producer发消息给Message Broker，不要求Message Broker对接收到的消息响应确认，Producer也不用关心Message Broker是否收到消息了。

b）Message Broker存储/转发阶段：

Message Broker必须提供持久性保障，并且每条消息在其消费队列里有唯一标识（这个唯一标识可以由Producer产生，也可以由Message Broker产生）。

c）Consumer消费阶段：

Consumer从Message Broker中获取到消息后，需要记录下消费的消息标识，以便在后续消费中防止对某个消息重复消费。比如Consumer获取到消息，消费完后，还没来得及从Message Broker删除消息，就挂了，这样Message Broker如果把消息重新加入待消费队列的话，那么这条消息就会被重复消费了。

2）Producer上产生的消息被Consumer仅消费一次，场景要求：

a）Producer发送消息到Message Broker阶段：

Producer发消息给Message Broker，Message Broker必须响应对消息的确认，并且Producer负责为该消息产生唯一标识，以防止Consumer重复消费（因为Producer发消息给Message Broker后，由于网络问题没收到Message Broker的响应，可能会重发消息给到Message Broker）。

b）Message Broker存储/转发阶段：

Message Broker必须提供持久性保障，并且每条消息在其消费队列里有唯一标识（这个唯一标识需要由Producer产生）。

c）Consumer消费阶段：

Consumer从Message Broker中获取到消息后，需要记录下消费的消息标识，以便在后续消费中防止对某个消息重复消费。比如Consumer获取到消息，消费完后，还没来得及从Message Broker删除消息，就挂了，这样Message Broker如果把消息重新加入待消费队列的话，那么这条消息就会被重复消费了。

实践Hello World

/// <summary>
/// 发送按钮事件
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
private void button1_Click(object sender, EventArgs e)
{if (!string.IsNullOrEmpty(textBox1.Text.Trim())){try{//判断私有队列是否存在if (!MessageQueue.Exists(@".\Private$\MyPrivateQueue")){//创建一个私有队列MessageQueue.Create(@".\Private$\MyPrivateQueue");}//实例一个队列var queue = new MessageQueue(@".\Private$\MyPrivateQueue");//发送消息（第一个参数为消息内容，第二个参数为消息标签或名称）queue.Send(textBox1.Text.Trim(), "TestLable");}catch (MessageQueueException ex){MessageBox.Show(ex.Message, "异常消息", MessageBoxButtons.OK, MessageBoxIcon.Error);}}
}

/// <summary>/// 接收按钮事件/// </summary>/// <param name="sender"></param>/// <param name="e"></param>private void button2_Click(object sender, EventArgs e){try{//判断私有队列是否存在if (!MessageQueue.Exists(@".\Private$\MyPrivateQueue")){//创建一个私有队列MessageQueue.Create(@".\Private$\MyPrivateQueue");}//实例一个队列var queue = new MessageQueue(@".\Private$\MyPrivateQueue");//读取消息，设置格式化方式queue.Formatter = new XmlMessageFormatter(new string[] { "System.String" });//读取第一个消息var message = queue.Receive();//显示消息内容label1.Text = message.Body.ToString();}catch (MessageQueueException ex){MessageBox.Show(ex.Message);}}

全文讲述framework中的消息队列的原理及简单的实践方法；更多framework的学习可以参考《framework精编手册》里面是有关framework的源码解析与技术点全家桶。点击即可前往

消息队列 好处或功能：

1、消息可以在断开连接的环境下发送。不需要同时运行正在发送和正在接收的应用程序。

2、使用快捷模式,消息可以非常快地发送。在快捷模式下,消息存储在内存中。

3、对于可恢复的机制,消息可以使用有保证的交付方式发送。可恢复的消`息存储在文件中。在服务器重新启动时发送它们。

4、用访问控制列表来保护消息队列,可以确定哪些用户可以发送或接收队列中的消息。消息还可以加密,避免网络嗅探器读取其中的数据。消息在发送时可以指定优先级,这样可以更快地处理高优先级的项。

5、Message Queuing 3.0支持多播消息的发送。

6、Message queuing 4.0支持病毒消息。病毒消息不能解析。可以定义病毒队列中不能解析的消息是可以移动的。例如,如果从正常的队列中读取消息后,对应作业要把消息插入数据库中,但消息不能插入数据库,因此该作业失败,该消息就会发送到病毒队列中。有人负责处理病毒队列,这个人应以能解析病毒消息的方式来处理该消息。

7、Message Queuing 5.0支持更安全学身份验证算法,可以处理大量队列(Message queuing 4.0在处理几千个队列时有性能问题)。