Linux系统编程—消息队列

Posted 2023-04-30

参考技术A 消息队列本质上是位于内核空间的链表，链表的每个节点都是一条消息。每一条消息都有自己的消息类型，消息类型用整数来表示，而且必须大于 0。每种类型的消息都被对应的链表所维护：

其中数字 1 表示类型为 1 的消息，数字2、3、4 类似。彩色块表示消息数据，它们被挂在对应类型的链表上。

值得注意的是，刚刚说过没有消息类型为 0 的消息，实际上，消息类型为 0 的链表记录了所有消息加入队列的顺序，其中红色箭头表示消息加入的顺序。

无论你是发送还是接收消息，消息的格式都必须按照规范来。简单的说，它一般长成下面这个样子：

所以，只要你保证首4字节(32 位 linux 下的 long)是一个整数就行了。
举个例子：

从上面可以看出，正文部分是什么数据类型都没关系，因为消息队列传递的是 2 进制数据，不一定非得是文本。

msgsnd 函数用于将数据发送到消息队列。如果该函数被信号打断，会设置 errno 为 EINTR。

参数 msqid：ipc 内核对象 id
参数 msgp：消息数据地址
参数 msgsz：消息正文部分的大小（不包含消息类型）
参数 msgflg：可选项
该值为 0：如果消息队列空间不够，msgsnd 会阻塞。
IPC_NOWAIT：直接返回，如果空间不够，会设置 errno 为 EAGIN.

返回值：0 表示成功，-1 失败并设置 errno。

msgrcv 函数从消息队列取出消息后，并将其从消息队列里删除。

参数 msqid：ipc 内核对象 id
参数 msgp：用来接收消息数据地址
参数 msgsz：消息正文部分的大小（不包含消息类型）
参数 msgtyp：指定获取哪种类型的消息

msgtyp = 0：获取消息队列中的第一条消息
msgtyp > 0：获取类型为 msgtyp 的第一条消息，除非指定了 msgflg 为MSG_EXCEPT，这表示获取除了 msgtyp 类型以外的第一条消息。
msgtyp < 0：获取类型 ≤|msgtyp|≤|msgtyp| 的第一条消息。
参数 msgflg：可选项。
如果为 0 表示没有消息就阻塞。
IPC_NOWAIT：如果指定类型的消息不存在就立即返回，同时设置 errno 为 ENOMSG
MSG_EXCEPT：仅用于 msgtyp > 0 的情况。表示获取类型不为 msgtyp 的消息
MSG_NOERROR：如果消息数据正文内容大于 msgsz，就将消息数据截断为 msgsz

程序 msg_send 和 msg_recv 分别用于向消息队列发送数据和接收数据。

msg_send 程序定义了一个结构体 Msg，消息正文部分是结构体 Person。该程序向消息队列发送了 10 条消息。
msg_send.c

程序 msg_send 第一次运行完后，内核中的消息队列大概像下面这样：

msg_recv 程序接收一个参数，表示接收哪种类型的消息。比如./msg_recv 4 表示接收类型为 4 的消息，并打印在屏幕。

先运行 msg_send，再运行 msg_recv。
接收所有消息

接收类型为 4 的消息

获取和设置消息队列的属性

msqid：消息队列标识符
cmd：控制指令
IPC_STAT:获得msgid的消息队列头数据到buf中
IPC_SET：设置消息队列的属性，要设置的属性需先存储在buf中，可设置的属性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes
buf：消息队列管理结构体。

返回值：
成功：0
出错：-1，错误原因存于error中
EACCESS：参数cmd为IPC_STAT，确无权限读取该消息队列
EFAULT：参数buf指向无效的内存地址
EIDRM：标识符为msqid的消息队列已被删除
EINVAL：无效的参数cmd或msqid
EPERM：参数cmd为IPC_SET或IPC_RMID，却无足够的权限执行

Linux编程:--消息队列（MessageQueue）相关概念和原理

 

一、消息队列概述

消息队列（MessageQueue，简称为MQ）
其本质是就是个队列，FIFO先进先出，只不过是队列中储放的主要内容是message，因而叫消息队列
主要用于：不同的服务server、进程process、线程thread相互间通信
二、选用消息队列的场景
①异步处理
②流量控制
③服务解耦
④发布订阅
⑤高并发缓冲

①异步处理
选用场景有短信提醒、终端状态推送、App推送、用户注册等
以秒杀系统为例：
如果不选用消息队列（同步处理）：假如选用同步处理，那 当客户成功消费之后，会先写进订单（例如在淘宝中生成一条订单信息），随后再通过短信提醒客户下单成功，最后再在系统中统计这条订单信息
假如选用消息队列（异步处理）：当客户成功消费之后，将信息写进消息队列，随后选用1种发布订阅模型，消息队列属于发布者，“订单、短信、统计”三者属于订阅者，这样，“订单、短信、统计”三者可以同时从消息队列中获取信息，就是1种异步处理的操作了.

优点： 更快速返回结果
减少等待，实现并发处理，提升系统总体性能

 

②流量控制（削峰）

使用消息队列隔离网关和后端服务，以达到流量控制和保护后端服务的目的
例如下图所示，在秒杀场景下：当多个用户通过APP抢购物品，那么就会同时向消息队列中写入数据，当消息队列中数据存满时，那么数据就不能再写入了，此时服务器就会给客户端APP回送一条类似于“淘宝双11时显示的系统繁忙，请稍后再试”的信息，从而达到流量控制

扩容的概念：通过上面我们知道，当消息队列写满之后，说明你的后端服务处理达到了极限，此时可以通过增加后端服务器的数量等来进行扩容，从而可以接收更多的服务器请求。如下图所示：

③服务解耦

使用消息队列实现系统的解耦
传统的发布订阅模式中：A系统负责数据分发，D系统、B系统、D系统分别来接收数据，因为不同系统的类型不同，因此A系统需要分别针对不同的系统调用其相关接口来对其进行服务。当有新系统加入时（例如下图的E系统），那么A系统就要修改源代码，来新增对E系统的接口实现服务。这样看来系统的耦合度太高

当使用消息队列的发布订阅模式之后：A系统直接将内容写入MQ中，然后其他系统也直接从MQ中读取数据，因为MQ的接口是统一的，因此大家使用一套就可以了。这样就使得系统的耦合性降低

 

④发布订阅
比如游戏里面跨服： 广播今天整体还剩多少把屠龙刀可以暴
广播用户暴的屠龙刀的消息

 

⑤高并发缓冲
日志服务（kafka在日志服务用的比较多）、监控上报

三、消息队列的相关概念和原理

Broker

Broker的概念来自与Apache ActiveMQ，通俗的讲就是MQ的服务器

 

消息的生产者、消费者

消息生产者（Producer）：发送消息到消息队列
消息消费者（Consumer）：从消息队列接收消息

 

点对点消息队列模型

消息生产者向一个特定的队列发送消息，消息消费者从该队列中接收消息
消息的生产者和消费者可以不同时处于运行状态。 每一个成功处理的消息都由消息消费者签收确认 （Acknowledge）

 

发布订阅消息模型-Topic

发布订阅消息模型中，支持向一个特定的主题Topic发布消息，0个或多个订阅者接收来自这个消息主题的消息
在这种模型下，发布者和订阅者彼此不知道对方。实际操作过程中，发布订阅消息模型中，支持向一个特定的主题Topic发布消 息，0个或多个订阅者接收来自这个消息主题的消息

RabbitMQ是点对点的消息队列模型，但是可以通过路由进行配置实现发布订阅。如下图所示：

 

消息的顺序性保证

基于Queue消息模型，利用FIFO先进先出的特性，可以保证消息的顺序性

 

消息的ACK确认机制

即消息的Ackownledge确认机制， 为了保证消息不丢失，消息队列提供了消息Acknowledge机制，即ACK机制：

当Consumer确认消息已经被消费处理，发送一个ACK给消息队列，此时消息队列便可以删除这个消息了
如果Consumer宕机/关闭，没有发送ACK，消息队列将认为这个消息没有被处理，会将这个消息重新发送给其他的Consumer重新消费处理

类似于TCP的ACK机制

 

消息的持久化

消息的持久化，对于一些关键的核心业务来说是非常重要的，启用消息持久化后， 消息队列宕机重启后，消息可以从持久化存储恢复，消息不丢失，可以继续消费处理
ZeroMQ不支持持久化，其他消息队列支持

 

消息的同步和异步收发

同步：消息的收发支持同步收发的方式
同时还有另一种同步方式：同步收发场景下，消息生产者和消费者双向应答模式，例如： 张三写封信送到邮局中转站，然后李四从中转站获得信，然后在写一份回执信，放到中转站，然后张三去取，当然张三写信的时候就得写明回信地址消息的接收如果以同步的方式（Pull）进行接收，如果队列中为空，此时接收将处于同步阻塞状态，会一直等待，直到消息的到达
异步：消息的收发同样支持异步方式：
异步发送消息，不需要等待消息队列的接收确认
异步接收消息，以Push的方式触发消息消费者接收消息

 

消息的事务支持

消息的收发处理支持事务
例如：在任务中心场景中，一次处理可能涉及多个消息的接收、处理，这处于同一个事务范围内，如果一个消息处理失败，事务回滚，消息重新回到队列中