【Linux 进程间通信（四）】System V 消息队列

上一篇博客我们讲了共享内存，接下来我们讲一下System V IPC的下一部分：消息队列

含义

原理

系统接口

mssget

作用

参数

返回值

msgsnd

作用

参数

返回值

msgrcv

作用

参数

返回值

msgctl

作用

参数

返回值

代码示例

server.cc(服务端)

client.cc(客户端)

comm.hpp(头文件)

效果演示

总结：

含义

消息队列允许不同进程之间通过在队列中发送和接收消息来进行通信。每个消息都有一个类型，接收进程可以选择接收特定类型的消息，也可以接收所有类型的消息。消息队列提供了一种异步的通信方式，发送进程可以将消息发送到队列中，而不需要等待接收进程准备好接收。

原理

消息队列创建：创建消息队列时，内核会分配一块内存区域来存储消息，并为该队列分配一个唯一的标识符。
发送消息：进程可以使用系统调用（如msgsnd）将消息发送到消息队列。发送消息时，进程需要指定目标队列的标识符、消息类型以及要发送的数据。
接收消息：进程可以使用系统调用（如msgrcv）从消息队列中接收消息。接收消息时，进程可以选择接收特定类型的消息，也可以接收队列中的下一条消息。
消息队列操作：除了发送和接收消息外，System V消息队列还提供了其他操作，如获取队列状态、删除队列等。
进程间同步：消息队列可以用于进程间的同步，比如一个进程等待另一个进程发送消息作为某种信号来进行同步操作。
消息类型：每个消息都有一个类型标识符，接收进程可以选择性地接收特定类型的消息，而不是仅仅接收队列中的下一条消息。
消息队列缓冲区：消息队列的内存由内核管理，因此可以确保高效的消息传递，并且能够处理大量消息。
持久性：与其他进程间通信机制相比，System V消息队列是持久的，即使发送消息的进程退出，消息队列仍然存在，直到显式删除为止。

系统接口

mssget

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

       int msgget(key_t key, int msgflg);

作用

msgget主要用来获取消息队列的标识符，如果消息队列不存在，可以使用msgflg参数中的IPC_CREAT来创建一个消息队列。

参数

key_t key:消息队列的键值，通常一个消息队列对应一个键值
int msgflg:用于指定创建消息队列时的权限标志。这个参数是一个位掩码，通常使用IPC_CREAT来创建一个新的队列，或者用IPC_EXCL | IPC_CREAT的结合确保指定key对应的消息队列不存在时再创建一个新的队列。

返回值

如果获取成功，该函数将返回key对应的消息队列的标识符，否则返回-1并将错误信息设置。

msgsnd

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

       int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

作用

msgsnd负责向消息队列中发送消息。

参数

int msqid:想要发送消息的消息队列标识符。
const void *msgp:msgp 参数是指向调用方定义的结构的指针，其常规形式如下：
struct msgbuf {
               long mtype;       /* message type, must be > 0 */
               char mtext[1];    /* message data */
           };
这个结构体需要用户自己进行创建。

mtype表示消息的类型，可以被设置为任何非负整数。将其设置成不同的值表示不同的消息类型或者优先级。

mtext是字符串，表示要发送消息的内容。

size_t msgsz:要发送数据的大小，通常用 sizeof(msgbuf)-sizeof(long) 计算。
int msgflg:可以定义msgsnd函数的行为，可以选择的标志位：0、IPC_NOWAIT、IPC_NOERROR、IPC_EXCEPT、MSG_NOERROR等。

返回值

如果发送成功，返回值为0，否则返回值为-1，并且设置错误信息。

msgrcv

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

       int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

作用

msgrcv负责向消息队列中读取信息。

参数

int msqid:想要发送消息的消息队列标识符。
const void *msgp:指向调用方定义的结构的指针
size_t msgsz:想要读取的消息的大小
int msgflg:可以定义msgrcv函数的行为，可以选择的标志位：0、IPC_NOWAIT、IPC_NOERROR、IPC_EXCEPT、MSG_NOERROR等。

返回值

如果读取成功，返回值为从mtext中读取到的字节数，否则返回-1，并设置错误信息。

msgctl

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

       int msgctl(int msqid, int cmd, struct msqid_ds *buf);

作用

msgctl函数用于控制控制消息队列，例如获取消息队列的状态、设置消息队列的属性以及删除消息队列。

参数

int msqid:想要发送消息的消息队列标识符。
int cmd:用于指定要执行的操作，可以选择：IPC_STAT、IPC_SET、IPC_RMID，分别是获取状态、设置属性、删除消息队列。
struct msqid_ds *buf: msqid_ds 是系统定义的结构体，可以直接创建，传入这个结构体用来获取msqid对应消息队列的状态或者设置属性，当cmd被设置为IPC_RMID时，这个参数可以忽略或者设置为0.

返回值

msgctl函数的返回值取决于执行的操作（cmd被设置成什么）以及是否出现错误：

如果执行成功，都是返回0
执行失败返回-1，并设置错误信息，可以通过errno进行查看。

代码示例

server.cc(服务端)

#include "comm.hpp"

int main()
{
    // 获取消息的结构体
    struct mymsg recept_message;
    // 打印消息队列信息的结构体
    struct msqid_ds ds;
    // 获取唯一键值
    key_t key = GetKey(pathname, proj_id);

    // 创建队列并获取队列ID
    int msgid = msgget(key, IPC_CREAT | 0666);
    if (msgid == -1)
    {
        std::cerr << "errno:" << errno << ",errstring:" << strerror(errno) << std::endl;
        return 1;
    }
    std::cout << msgid << std::endl;

    // 打印队列属性
    if (msgctl(msgid, IPC_STAT, &ds) == -1)
    {
        std::cerr << "errno:" << errno << ",errstring:" << strerror(errno) << std::endl;
        return 1;
    }
    std::cout << "消息队列属性如下：";
    std::cout << "所有者的UID: " << ds.msg_perm.uid << std::endl;
    std::cout << "所有者的GID:" << ds.msg_perm.gid << std::endl;
    std::cout << "读写权限： " << ds.msg_perm.mode << std::endl;
    std::cout << "当前消息队列中的消息个数 " << ds.msg_qnum << std::endl;
    std::cout << "消息队列的最大尺寸： " << ds.msg_qbytes << std::endl;

    // 服务端读取消息
    if (msgrcv(msgid, &recept_message, sizeof(recept_message) - sizeof(long), 1, 0) == -1)
    {
        std::cerr << "errno:" << errno << ",errstring:" << strerror(errno) << std::endl;
        return 1;
    }
    std::cout << "服务端收到消息：" << recept_message.s << std::endl;

    std::cout << "准备关闭消息队列" << std::endl;
    sleep(5);

    // 关闭队列
    msgctl(msgid, IPC_RMID, 0);
    std::cout << "已关闭消息队列" << std::endl;

    return 0;
}

client.cc(客户端)

#include "comm.hpp"

int main()
{
    struct mymsg send_message;
    // 获取消息队列标识符
    key_t key = GetKey(pathname, proj_id);
    // DEBUG
    std::cout << "获取消息队列标识符" << std::endl;
    int msgid = msgget(key, IPC_NOWAIT);

    std::cout << "msgid:" << msgid << std::endl;
    // 向消息队列中发送消息 

    strcpy(send_message.s, "客户端发送的消息");
    if (msgsnd(msgid, &send_message, sizeof(send_message) - sizeof(long), 0) == -1)
    {
        std::cerr << "errno:" << errno << ",errstring:" << strerror(errno) << std::endl;
        return 1;
    }

    return 0;
}

comm.hpp(头文件)

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


#define SIZE 1024
const char *pathname = "/home/gpc/pipeline/message_queue";
const int proj_id = 0x11223344;

struct mymsg
{
    long mtype = 1;
    char s[SIZE];
};

key_t GetKey(const char *pathname,const int proj_id)
{
    key_t key = ftok(pathname,proj_id);
    if(key < 0)
    {
        std::cerr<<"errno:"<<errno<<",errstring:"<<strerror(errno)<<std::endl;
        return 1;
    }
    return key;
}

效果演示

消息队列演示效果

总结：

Ⅰ、 System V 消息队列是通过标识符引用的，而不是文件描述符。这意味着各种基于文件描述符的I/O技术将无法应用于消息队列上。
Ⅱ、使用键key而不是文件名来标识消息队列会增加额外的程序设计复杂性，同时还需要使用ipcs和ipcrm来替换ls和rm。ftok()函数通常能产生一个唯一的键，但却无法百分百保证。使用IPC_PRIVATE能确保产生唯一的队列标识符，但需要使这个标识符对需要用到它的其他进程可见。
Ⅲ、消息队列是无法连接的，内核不会像对待管道、FIFO以及socket那样维护引用队列的进程数。因此，就需要注意以下问题：