Linux系统编程系列之进程间通信-消息队列

news2024/11/27 3:42:29

一、什么是消息队列

消息队列是system-V三种IPC对象之一，是进程间通信的一种方式。

二、消息队列的特性

允许发送的数据携带类型（指定发送给谁），具有相同类型的数据在消息队列内部排队，读取的时候也要指定类型，然后依次读出数据。可以理解为消息队列就是一个多管道集合。

三、消息使用场景

由于每个消息都携带有类型，相同的类型自成一队，因此读取方向可以根据类型来挑选不同的队列，所以说MSG适用于“多对一”的场景。

（1）系统日志。

（2）多个不同的，不相关的进程向同一管道输入数据。

四、函数API接口

1、指定消息队列功能

// 指定消息队列功能
int msgget(key_t key, int msgflg);

// 接口说明：
    返回值：消息队列MSG对象ID
    参数key：键值，全局唯一标识，可由ftok()产生
    参数msgflg：操作模式与读写权限，与文件操作函数open类似。

2、向MSG对象发送消息

// 向MSG对象发送消息
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

// 接口说明
    msggid：MSG对象的ID，由msgget()获取
    msgp：一个指向等待被发送的消息的指针，由于MSG中的消息最大的特点是必须有一个整数标识，用以区分MSG中的不同的消息，因此MSG的消息会使用一个特别的结构体来表达:

struct msgbuf
{
    long mtype;    // 消息类型固定，必须大于0
    char text[100];    // 消息正文（可变，自己定义）
    
};

    msgsz：消息正文的长度（单位字节），注意不含类型长度
    msgflg：发送选项
    0：默认发送模式，在MSG缓冲区已满的情形下阻塞，直到缓冲区变为可用状态
    IPC_NOWAIT：非阻塞发送模式，在MSF缓冲区已满的情形下直接退出函数并设置错误码为EAGAIN

3、从MSG对象接收消息

// 从MSG对象接收消息
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

// 接口说明
    msgqid：MSG对象ID，由msgget()获取
    msgp：存放消息的内存入口
    msgsz：存放消息的内存大小
    msgtyp：欲接收消息的类型
        0：不区分类型，自己读取MSG中的第一个消息
        大于0：读取类型为指定msgtyp的第一个消息（若msgfla被配置了MSG_EXCEPT则读取除了类型msgtyp的第一个消息）
        小于0：读取类型小于等于msgtyp绝对值的第一个具有最小类型的消息。
    msgflg：接收选项
        0：默认接收模式，在MSG中无指定类型消息时阻塞
        IPC_NOWAIT：非阻塞接收模式，在MSG中无指定类型消息时直接退出函数并设置错误码为ENOMSG
        MSG_EXCEPT：读取出msgtyp之外的第一个消息
        MSG_NOERROR：如果待读取的消息尺寸比msgsz大，直接切割消息并返回msgsz部分，读不了的部分直接丢弃。若没有设置该项，则函数将出错返回并设置错误码为E2BIG

4、删除消息队列对象

在前一篇博客IPC对象中有讲，可以使用命令来删除，同时也可以采用函数的方式来删除

// 删除MSG对象
int msgctl(int msqid, int cmd, struct msgqid_ds *buf);

// 接口说明
msqid：MSG对象ID
cmd：控制命令字
    IPC_STAT：获取该MSG的信息，储存在结构体msgqid_ds中
    IPC_SET： 设置该MSG的信息，储存在结构体msgqid_ds中
    IPC_RMID：立即删除该MSG，并且唤醒所有阻塞在该MSG上的进程，同时忽略第三个参数

五、消息队列使用步骤

1、使用ftok()获取key值

2、调用msgget()，指定为消息队列

3、调用msgsnd()或者msgrcv()来发送或者接收信息。

4、使用命令或者函数删除消息队列（可选）。

六、案例

使用消息队列来实现两个进程间进行相互收发信息

// 消息队列的案例

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

// 编译版本，一个直接编译，另外一个把0x01和0x02位置互换后编译
#define SEND_MTYPE 0x01
#define RECV_MTYPE 0x02

struct msgbuf 
{
    long mtype;      // 消息类型，固定的，必须大于0 
    char mtext[32];  // 可变的，自己定义
};


int main(int argc, char *argv[])
{
    // 1、获得KEY值
    key_t msg_key = ftok("./", 1);
    if(msg_key == -1)
    {
        perror("ftok fail");
        return -1;
    }

    // 2、指定消息队列功能
    int msg_id = msgget(msg_key, IPC_CREAT | 0666);
    if(msg_id == -1)
    {
        perror("msgget fail");
        return -1;
    }

    // 3、消息队列的读写操作
    struct msgbuf Buf = {0};
    int ret = 0;
    pid_t pid = fork();
    // 父进程
    if(pid > 0)
    {   
        Buf.mtype = SEND_MTYPE; // 指定消息类型
        while(1)
        {
            printf("please input data:\n");
            // fgets("%s", Buf.mtext);
            fgets(Buf.mtext, sizeof(Buf.mtext), stdin);
            // 发送消息
            ret = msgsnd(msg_id, &Buf, sizeof(Buf.mtext), 0);
            if(ret == -1)
            {
                perror("msgsnd fail");
            }
            else if(ret == 0)
            {
                printf("send data success: %s\n", Buf.mtext);
            }
            memset(Buf.mtext, 0, sizeof(Buf.mtext));
        }
    }
    // 子进程
    else if(pid == 0)
    {
        Buf.mtype = RECV_MTYPE; // 指定消息类型
        while(1)
        {
            // 接收消息
            ret = msgrcv(msg_id, &Buf, sizeof(Buf.mtext), Buf.mtype, 0);
            if(ret > 0)
            {
                printf("read data success: %s\n", Buf.mtext);
            }
        }
    }
    else
    {
        perror("fork fail");
        return -1;
    }

    return 0;
}