文章/答案/技术大牛

发布

社区首页 >专栏 >【Linux系统】进程间通信-System V消息队列

【Linux系统】进程间通信-System V消息队列

用户11396661

发布于 2025-04-03 00:50:44

11800

代码可运行

文章被收录于专栏：C++开发C++开发

运行总次数：0

代码可运行

1.进程间通信（IPC：Inter-Process Communication）

常见的IPC方式：

匿名管道，命名管道，共享内存，消息队列，信号量。

System -V 消息队列

生命周期随内核，不随进程。

也就是说，当前面一个创建了一个消息队列，但是没有进行销毁，最后进程也结束了。当下次新的进程再次创建的时候，就会出错。

消息队列的销毁：1.命名销毁。 2.msgctl。 3.系统重启。

需要注意的是，如果在编程过程中，虽然写了msgctl进行销毁，如果提前终止了进程，消息队列也是没有进行销毁的。

下面会对这些函数进行封装。包括msgqueue类，客户端，服务端。

2.MSGGET函数

函数原型：

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

int msgget(key_t key, int msgflg);

成功返回消息队列唯一标识符。

错误返回-1。

功能：

官方说明：获取一个System V 消息队列标识符（get a System V message queue identifier）。类型为int

也就是说，通过这个函数，可以创建一个消息队列，然后返回一个消息队列标识符（这个是给用户态唯一确定的标识符，在内核态是通过key来唯一定位一个标识符）。

参数key

The msgget() system call returns the System V message queue identifier associated with the value of the key argument.

msgget通过key去创建消息队列，key值我们可以通过ftok函数去创建。（下面会讲解）。

参数msgflag

IPC_CREATE：如果key对应的消息队列不存在则创建，如果存在就获取它的ID。

IPC_CREATE|IPC_EXCL_：两个一起使用，唯一的不同就是如果存在则出错。

另外可以加上权限：

如果权限设置为0666，系统的umask为022，最终的权限就是0666&(~022)=0644.

权限是八进制的，打印的时候，要记得按八进制打印更方便观察。第1位是特殊标记位，未启用。

3.ftok函数

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

key_t ftok(const char *pathname, int proj_id);

pathname是文件路径，proj_id的值是0-255的值（也就是取int的低8位）。

然后回通过pathname和proj_id唯一创建一个key_t类型的键值。

如果创建失败，可以更换文件路径或者proj_id进行重新尝试。

3.msgctl函数

msgctl函数用于对System V消息队列进行控制操作，如获取消息队列的状态、设置消息队列的属性或删除消息队列。

函数原型

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

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

参数msqid

这个就是上面msgget函数的返回值，消息队列的标识符。

参数cmd

这个就是要进行哪种操作。

IPC_STAT	获取消息队列的状态信息
IPC_SET	设置属性
IPC_RMID	删除消息队列

buf：指向msqid_ds结构体的指针，用于获取或设置消息队列的信息。对于IPC_STAT和IPC_SET，buf必须是一个有效的指针；对于IPC_RMID，buf可以是NULL。

struct msqid_ds {
    struct ipc_perm msg_perm;    // 权限信息
    struct msg *msg_first;       // 消息队列中的第一个消息
    struct msg *msg_last;        // 消息队列中的最后一个消息
    ulong msg_cbytes;            // 消息队列中当前的总字节数
    ulong msg_qnum;              // 消息队列中当前的总消息数
    ulong msg_qbytes;            // 消息队列中最大可容纳的字节数
    pid_t msg_lspid;             // 最近一个执行`msgsnd`的进程PID
    pid_t msg_lrpid;             // 最近一个执行`msgrcv`的进程PID
    time_t msg_stime;            // 最近一次执行`msgsnd`的时间
    time_t msg_rtime;            // 最近一次执行`msgrcv`的时间
    time_t msg_ctime;            // 最近一次改变消息队列的时间
};

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


int main()
{
    key_t key=ftok("./",123);
    int msgid=msgget(key,IPC_CREAT|0666);
    if(-1==msgid){
        std::cout<<"创建消息队列失败！"<<std::endl;
    }
    struct msqid_ds ds;
    msgctl(msgid,IPC_STAT,&ds);

    printf("Mode: %04o\n", ds.msg_perm.mode);
    std::cout<<std::setw(4)<<std::setfill('0')<<std::oct<<ds.msg_perm.mode<<std::endl;
    
    getchar();
    msgctl(msgid,IPC_RMID,0);
    return 0;
}

删除消息队列

bool destroy(){
    int n=msgctl(_msgfd,IPC_RMID,0);
    if(-1==n){
        std::cerr<<"msgctl error"<<std::endl;
        return false;
    }
    std::cout<<"destroy success"<<std::endl;
    return true;
}

4.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);

msqid参数

消息队列唯一标识符

msgp参数

必须满足第一个为long类型，其他的是数据部分。可以设置为如下类型：

struct mymsg {
    long mtype;  // 消息类型
    char mtext[1024];  // 消息数据
};

msgsz参数

这是消息数据的大小，不包括long类型的mtype大小。

msgflag参数

IPC_NOWAIT：如果消息队列已满，则立即返回，而不是阻塞等待。

0：默认行为，如果消息队列已满，则阻塞等待，直到消息被发送或超时。

5.msgrcv函数

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

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);

msgid，msgp，msgsz和上面的一样。

msgtyp参数

消息的类型，用于选择性接收消息：

0	收队列中第一个消息。
正整数	接收类型等于 msgtyp 的消息
负整数	接收类型小于或等于 abs(msgtyp) 的消息

通过这个参数，可以设置优先级，看先读取哪个，如果失败，再读取哪些。

msgflag参数

IPC_NOWAIT：如果队列中没有符合条件的消息，则立即返回，而不是阻塞等待。

MSG_NOERROR：如果消息的大小超过 msgsz，则截断消息（只接收前 msgsz 字节）。

返回值

成功：返回接收到的消息数据部分的大小（以字节为单位）。

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

MsgQueue类封装

对于客户端，只进行消息队列的获取，不进行创建，而服务端确确实实要进行创建一个新的消息队列。服务端msgget就设置参数为IPC_CREATE|IPC_EXCL|0666。

#pragma once

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

#define PATH_NAME "/tmp"
#define PROJ_ID 0x12
#define GLOBAL_MSG_FD -1
#define MTEXT_SIZE 128
#define CLIEN_TYPE 1
#define SERVER_TYPE 2

namespace msgqueue{
    class MsgQueue{
        public:
            //定义类型结构体
            struct MsgBuff{
                long mtype;
                char mtext[MTEXT_SIZE];
            };
    
            MsgQueue():_msgfd(GLOBAL_MSG_FD){}
            //选项由外部决定
            bool create(int flag){
                key_t key=ftok(PATH_NAME,PROJ_ID);
                if(-1==key){
                    std::cerr<<"key error"<<std::endl;
                    return false;
                }
                _msgfd=msgget(key,flag);
                if(-1==_msgfd){
                    std::cerr<<"msgget error"<<std::endl;
                    return false;
                }
                std::cout<<"create success!msgid:"<<_msgfd<<" key:"<<key<<std::endl;
                return true;
            }
            bool destroy(){
                int n=msgctl(_msgfd,IPC_RMID,0);
                if(-1==n){
                    std::cerr<<"msgctl error"<<std::endl;
                    return false;
                }
                std::cout<<"destroy success"<<std::endl;
                return true;
            }
            bool send(int mtype,std::string& mtext){
                MsgBuff msgbuff;
                memset(&msgbuff,0,sizeof(msgbuff));
                msgbuff.mtype=mtype; 
                memcpy(&msgbuff.mtext,mtext.c_str(),mtext.size());    
                //发送数据
                int n=msgsnd(_msgfd,&msgbuff,mtext.size(),0);
                if(-1==n){
                    std::cerr<<"msgsnd error"<<std::endl;
                    return false;
                }
                return true;
            }
    
            bool recv(int mtype,std::string& mtext){
                MsgBuff msgbuff;
                memset(msgbuff.mtext,0,sizeof(msgbuff.mtext));
                size_t end=msgrcv(_msgfd,&msgbuff,MTEXT_SIZE-1,mtype,0);
                if(-1==end){
                    std::cerr<<"msgrcv error"<<std::endl;
                    return false;
                }
                mtext=msgbuff.mtext;
                return true;
            }
        private:
            int _msgfd;
    };

    class Server:public MsgQueue{
        public:
            Server(){
                if(false==create(IPC_CREAT|IPC_EXCL|0666)){
                    std::cerr<<"create error"<<std::endl;
                }
            }
            ~Server(){
                if(false==destroy()){
                    std::cerr<<"destroy error"<<std::endl;
                }
            }

    };

    class Client:public MsgQueue{
        public:
            Client(){
                //只进行获取，不进行创建，若干不存在，则出错
                if(false==create(IPC_CREAT)){
                    std::cerr<<"create error"<<std::endl;
                }
            }
            ~Client(){
            }
    };
}

客户端：

#include "msgqueue.hpp"

int main()
{
    msgqueue::Client client;
    while(1){
        std::string s;
        std::cout<<"请输入发送的信息：";
        std::cin>>s;
        client.send(CLIEN_TYPE,s);
        if("quit"==s){
            break;
        }
    } 
}

服务端：

#include "msgqueue.hpp"

int main()
{
    msgqueue::Server server;
    while(1){
        std::string s;
        server.recv(CLIEN_TYPE,s);
        if("quit"==s){
            break;
        }
        std::cout<<s<<std::endl;
    }
}