【Linux-进程间通信】消息队列

消息队列的引入以及基本概念

在Linux操作系统中，消息队列（Message Queue）是一种进程间通信（Inter-process Communication, IPC）机制，允许一个进程向另一个进程发送和接收消息。消息队列在操作系统内核中维护，允许消息以队列的形式存储，并且可以在发送和接收之间进行排队。这种机制在需要多个进程协作时非常有用，比如在客户端和服务器之间通信。

消息队列的引入主要是为了实现解耦、异步处理、削峰填谷、容错性以及系统的高可用性。在现代的分布式系统中，服务之间的直接依赖可能会导致系统复杂度增加，尤其是在面对高并发、大数据量处理时。消息队列能够作为中间层，通过在生产者和消费者之间插入一个队列，让系统各个模块之间相互独立、松耦合，从而提高系统的稳定性和性能。

1. 解耦

在传统系统中，模块A可能会直接调用模块B的接口并等待其处理结果。这种强依赖增加了系统的耦合性，使得模块之间的修改和升级变得困难。通过引入消息队列，模块A可以将消息放入队列，不需要知道B是如何处理的，B也可以独立地从队列中消费消息并处理。

2. 异步处理

很多场景下，生产者需要处理大量请求，如果每次请求都需要等待消费者完成处理，系统响应时间将被拉长。消息队列允许生产者将消息发送到队列中，然后立即返回，由消费者在后台异步处理消息，极大提升了系统的响应速度。

3. 削峰填谷

在面对流量激增的场景中，系统可能会出现瞬时的负载过高，导致宕机或服务不可用。通过消息队列，生产者可以将大量的请求写入队列，而消费者可以根据自身的处理能力从队列中读取消息。即使系统在某一时刻产生了大量消息，消费者也可以有序处理，不至于崩溃。

4. 容错性

消息队列通常支持消息的持久化和重试机制，即使在消费者处理消息时出现异常，也可以确保消息不会丢失。在消费者恢复后，它可以继续从队列中消费未处理的消息，保障系统的容错性。

消息队列的基本概念

1.消息：消息是消息队列中传递的基本单位，通常包含数据的负载和一些元数据，如消息的唯一ID、时间戳等。消息可以是字符串、JSON、XML或者任何其他格式的数据。

2.生产者：生产者是发送消息到队列的一方，它将消息按照指定的格式写入消息队列。生产者不需要关心消息的处理方式或处理时间，因而可以专注于业务逻辑。

3.消费者：消费者是从消息队列中接收消息并进行处理的一方。消费者可以是单个进程或多个进程，并且可以同时处理来自多个生产者的消息。

4.队列：队列是消息的存储结构，通常是先进先出（FIFO）的，但有时可以根据优先级或其他规则处理消息。队列中的消息可以存储一定时间，直到被消费或者过期。

5.Broker：消息中介（Broker）是负责管理消息的传递与存储的服务，它在生产者和消费者之间充当中间层，接收消息并确保它们被正确发送到目标消费者。常见的消息中介包括RabbitMQ、Kafka、ActiveMQ等。

消息队列与命名管道和共享内存的不同

消息队列的原理

消息队列的核心思想是将消息存储在一个有序队列中，发送进程将消息放入队列，接收进程从队列中读取消息。消息队列由操作系统内核维护，消息按照顺序或者优先级进行存放。消息队列的设计使得不同进程不需要同时存在就可以实现消息传递，这是异步通信的基础。

 系统中可能存在大量的消息队列，内核需要使用如下数据结构对消息队列进行管理

消息队列工作流程

•生产者进程：通过msgsnd()将消息放入队列，并指定消息类型。如果消息队列已满，生产者进程可以选择等待队列有空位或直接返回错误。

•消费者进程：通过msgrcv()从队列中读取消息，并指定需要的消息类型。如果没有符合条件的消息，消费者可以选择阻塞等待消息的到来或直接返回错误。

•消息传递机制：消息队列在内核中排队，确保消息按照指定的顺序或优先级传递给接收者。

System V 消息队列的主要函数

msgget

msgget 是一个在 POSIX 系统中用于消息队列操作的函数。该函数用于获取一个消息队列的标识符，或者创建一个新的消息队列。

参数：

• key: 这是一个整型值，通常由 ftok 函数生成，用于唯一标识一个消息队列。如果这个值设置为 IPC_PRIVATE，则创建一个只能由调用进程访问的消息队列。

• msgflg: 这是一个位掩码，用于指定创建消息队列时的权限和标志。它可以是以下值的按位或操作组合：

返回值：

• 成功时，返回消息队列的标识符（一个非负整数）。

• 失败时，返回 -1，并设置 errno 以指示错误。

#include <iostream>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int main()
{
    key_t key = ftok("/home/wuxu/day27/func", 66);
    int msgid = msgget(key, IPC_CREAT | 0666);

    if (msgid == -1)
    {
        perror("msgget");
        exit(1);
    }
    return 0;
}

可以借助ipcs -q查看系统中存在的消息队列

msgrcv

msgsnd 函数用于将一条消息发送到指定的消息队列。

参数：

• msqid: 消息队列标识符，由 msgget 函数返回。

• msgp: 指向消息结构的指针。消息结构通常定义如下：

  struct msgbuf {
      long mtype;       /* 消息类型 */
      char mtext[1];    /* 消息数据 */
  };

注意，mtext 字段的大小应该至少为 msgsz，并且结构体的大小应该足够大以包含 mtext 中的数据。

• msgsz: msgp 中 mtext 字段的大小，即要发送的数据的实际字节数。

• msgflg: 控制消息发送的标志。如果设置为 0，msgsnd 将阻塞直到消息队列有足够的空间来放置消息。可以设置的标志包括：IPC_NOWAIT: 如果消息队列已满，msgsnd 将不等待，而是立即返回错误。

返回值：

• 成功时返回 0。

• 失败时返回 -1，并设置 errno 以指示错误。

msgsnd

msgrcv 函数用于从指定的消息队列中接收一条消息。

参数：

• msgsz: msgp 中 mtext 字段的最大大小，即可以接收的数据的最大字节数。

• msgtyp: 指定要接收的消息类型。可以是以下之一：

• msgflg: 控制消息接收的标志

返回值：

• 成功时返回实际读取的消息字节数。

• 失败时返回 -1，并设置 errno 以指示错误。

msgctl

msgctl是POSIX系统调用之一，用于控制消息队列的行为。它允许你获取消息队列的当前状态、设置消息队列的属性，或者删除消息队列

参数说明：

• msqid：消息队列的标识符，由msgget函数返回。

• cmd：要执行的命令

• buf：指向msqid_ds结构体的指针，该结构体用于保存消息队列的属性或状态信息。如果cmd参数是IPC_RMID，则此参数可以为NULL。

返回值：

• 0：成功。如果函数调用成功，msgctl返回0。

• -1：失败。如果函数调用失败，msgctl返回-1，并且设置全局变量errno来指示错误类型。可以使用perror或strerror函数来获取与errno相关的错误信息。

msqid_ds结构体定义在<sys/msg.h>头文件中，它包含以下字段：

 【示例】获取消息队列的状态

#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

int main() {
    key_t key;
    int msgid;
    struct msqid_ds msq_ds;
    int result;

    // 生成key
    key = ftok("/home/wuxu/day28/func", 5);

    // 获取消息队列标识符
    msgid = msgget(key, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget failed");
        exit(1);
    }

    // 获取消息队列状态
    result = msgctl(msgid, IPC_STAT, &msq_ds);
    if (result == -1) {
        perror("msgctl failed");
        exit(1);
    }

    printf("Message queue status:\n");
    printf("msg_perm.uid = %d\n", msq_ds.msg_perm.uid);
    printf("msg_perm.gid = %d\n", msq_ds.msg_perm.gid);
    printf("msg_qnum = %ld\n", msq_ds.msg_qnum);
    printf("msg_qbytes = %ld\n", msq_ds.msg_qbytes);
    printf("msg_lspid = %ld\n", msq_ds.msg_lspid);
    printf("msg_lrpid = %ld\n", msq_ds.msg_lrpid);
    printf("msg_stime = %ld\n", msq_ds.msg_stime);
    printf("msg_rtime = %ld\n", msq_ds.msg_rtime);
    printf("msg_ctime = %ld\n", msq_ds.msg_ctime);

    return 0;
}

设置消息队列状态

#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>

int main()
{
    sleep(2);
    key_t key;
    int msgid;
    struct msqid_ds msq_ds;
    int result;

    // 生成key
    key = ftok("/home/wuxu/day28/func", 65);

    // 获取消息队列标识符
    msgid = msgget(key, 0666 | IPC_CREAT);
    if (msgid == -1)
    {
        perror("msgget failed");
        exit(1);
    }

    // 设置消息队列属性
    memset(&msq_ds, 0, sizeof(msq_ds));
    msq_ds.msg_perm.uid = getuid(); // 设置所有者为当前用户
    msq_ds.msg_perm.gid = getgid(); // 设置组为当前用户组
    msq_ds.msg_qbytes = 1024;       // 设置队列最大长度为1024字节

    result = msgctl(msgid, IPC_SET, &msq_ds);
    if (result == -1)
    {
        perror("msgctl failed");
        exit(1);
    }
    
    printf("Message queue attributes set successfully.\n");

    return 0;
}