Linux进程间通信(IPC)编程实践（四）Posix消息队列--基本API的使用

缺乏、安全感 2022-05-14 13:40 443阅读 0赞

# 1 概述 #

## 1.1 posix消息队列与system v消息队列的差别 ##

POSIX表示可移植操作系统接口 ( Portable Operating System Interface of UNIX，缩写为 POSIX )。

(1) 对posix消息队列的读总是返回最高优先级的最早消息，对system v消息队列的读则可以返回任意指定优先级的消息。

(2) 当往一个空队列放置一个消息时，posix消息队列允许产生一个信号或启动一个线程，system v消息队列则不提供类似机制。

## 1.2 队列中的每个消息具有如下属性 ##

(1) 一个无符号整数优先级(posix)或一个长整数类型(system v)  
(2) 消息的数据部分长度(可以为0)  
(3) 数据本身(如果长度大于0)

#    2 Posix消息队列操作函数 #

## 2.1 创建/获取一个消息队列 ##

mqd_t mq_open(const char *name, int oflag); //专用于打开一个消息队列  
    mqd_t mq_open(const char *name, int oflag, mode_t mode,  
                  struct mq_attr *attr);

(1) 参数

(a) name:  消息队列名字;

(b) oflag: open函数类型, 可以按位或

O\_CREAT 没有该对象则创建  
   O\_EXCL 如果O\_CREAT指定，但name不存在，就返回错误  
   O\_NONBLOCK 以非阻塞方式打开消息队列  
   O\_RDONLY 只读  
   O\_RDWR 读写  
   O\_WRONLY 只写

S\_IWUSR 用户/属主写  
   S\_IRUSR 用户/属主读  
   S\_IWGRP 组成员写  
   S\_IRGRP 组成员读  
   S\_IWOTH 其他用户写  
   S\_IROTH 其他用户读

(d) attr: 指定消息队列的属性;

(2) 返回值:

成功: 返回消息队列文件描述符;

失败: 返回-1;

(3) 注意-Posix IPC名字限制:

(a) 必须以"/"开头, 并且后面不能还有"/", 形如:/file-name;  
(b) 名字长度不能超过NAME\_MAX  
(c) 链接时:Link with -lrt（Makefile中使用实时链接库-lrt）

类似对文件的open，我们可以用mq\_open来打开一个已经创建的消息队列或者创建一个消息队列。这个函数返回一个叫做mqd\_t类型的返回值，其本质上还是一个文件描述符，只是在这这里被叫做消息队列描述符（message queue descriptor），在进程里使用这个描述符对消息队列进程操作。所有被创建出来的消息队列在系统中都有一个文件与之对应，这个文件名是通过name参数指定的，这里需要注意的是：name必须是一个以”/”开头的字符串，比如我想让消息队列的名字叫”message”，那么name应该给的是”/message”。消息队列创建完毕后，会在/dev/mqueue目录下产生一个以name命名的文件，我们还可以通过cat这个文件来看这个消息队列的一些状态信息。其它进程在消息队列已经存在的情况下就可以通过mp\_open打开名为name的消息队列来访问它。

[root@localhost mqueue]# cat /dev/mqueue/test 
    QSIZE:48         NOTIFY:0     SIGNO:10    NOTIFY_PID:4442  
    [root@localhost mqueue]#

## 2.2 关闭一个消息队列 ##

#include <mqueue.h>
    int mq_close(mqd_t mqdes);

返回： 成功时为0，出错时为-1。  
功能： 关闭已打开的消息队列。

注意：System V没有此功能函数调用

## 2.3 删除一个消息队列 ##

int mq_unlink(const char *name);

/\*\*   
System V 消息队列   
通过msgctl函数, 并将cmd指定为IPC\_RMID来实现   
int msgctl(int msqid, int cmd, struct msqid\_ds \*buf);   
\*\*/

返回： 成功时为0，出错时为－1  
功能： 从系统中删除消息队列。

### 2.3.1 mqueue\_test.cpp ###

#include <stdio.h>
    #include <mqueue.h>
    
    int main()  
    {  
        mqd_t mqid = mq_open("/abc", O_CREAT|O_RDONLY, 0666, NULL);  
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
        
        printf("mq_open success!\n");
        
        mq_close(mqid);  
        mq_unlink("/abc");
        
        printf("unlink success!\n");
        
        return 0;
    }

编译

g++ -o mqueue_test ./mqueue_test.cpp -lrt

## 2.4 获取/设置消息队列属性 ##

#include <mqueue.h>
    int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
    int mq_setattr(mqd_t mqdes, const struct mq_attr *attr, struct mq_attr *attr);

均返回：成功时为0， 出错时为-1

参数:

newattr: 需要设置的属性

oldattr: 原来的属性

每个消息队列有四个属性：

struct mq_attr
    {
        long mq_flags;      /* message queue flag : 0, O_NONBLOCK */
        long mq_maxmsg;     /* max number of messages allowed on queue*/
        long mq_msgsize;    /* max size of a message (in bytes)*/
        long mq_curmsgs;    /* number of messages currently on queue */
    };

### 2.4.1 mqueue\_get.cpp ###

#include <stdio.h>
    #include <mqueue.h>
    
    
    int main(int argc,char **argv)  
    {  
        mqd_t mqid = mq_open("/test", O_RDONLY|O_CREAT, 0666, NULL);
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
      
        struct mq_attr attr;  
        if (mq_getattr(mqid, &attr) == -1)
        {
            printf("mq_getattr error!\n");
            return -1;
        }
        
        printf("Max messages on queue: %ld\n", attr.mq_maxmsg);  
        printf("Max message size: %ld\n", attr.mq_msgsize);  
        printf("current messages: %ld\n", attr.mq_curmsgs);  
      
        mq_close(mqid);
        
        return 0;  
    }

对比System V：  
通过msgctl函数, 并将cmd指定为IPC\_STAT/IPC\_SET来实现

int msgctl(int msqid, int cmd, struct msqid\_ds \*buf);

另外每个消息均有一个优先级，它是一个小于MQ\_PRIO\_MAX的无符号整数  
\#define MQ\_PRIO\_MAX 32768

## 2.5 发送消息/读取消息 ##

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);
    
    int mq_timedsend(mqd_t mqdes, const char *msg_ptr, size_t msg_len, 
    unsigned int msg_prio, const struct timespec *abs_timeout);
    
    ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);
    
    ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr, size_t msg_len, 
    unsigned int *msg_prio, const struct timespec *abs_timeout);

返回：成功时为0，出错为－1

返回：成功时为消息中的字节数，出错为－1  
参数：

msg\_prio 消息的优先级

abs\_timeout 超时时间

消息队列的限制：  
MQ\_OPEN\_MAX : 一个进程能够同时拥有的打开着消息队列的最大数目  
MQ\_PRIO\_MAX : 任意消息的最大优先级值加1

在一个消息队列创建完毕之后，我们可以使用mq\_send来对消息队列发送消息，mq\_receive来对消息队列接收消息。正常的发送消息一般不会阻塞，除非消息队列处在某种异常状态或者消息队列已满的时候，而消息队列在空的时候，如果使用mq\_receive去试图接受消息的行为也会被阻塞，所以就有必要为两个方法提供一个带超时时间的版本。这里要注意的是msg\_prio这个参数，是用来指定消息优先级的。每个消息都有一个优先级，取值范围是0到sysconf(\_SC\_MQ\_PRIO\_MAX) – 1的大小。在Linux上，这个值为32768。默认情况下，消息队列会先按照优先级进行排序，就是msg\_prio这个值越大的越先出队列。同一个优先级的消息按照fifo原则处理。在mq\_receive方法中的msg\_prio是一个指向int的地址，它并不是用来指定取的消息是哪个优先级的，而是会将相关消息的优先级取出来放到相关变量中，以便用户自己处理优先级。

### 2.5.1 mqueue\_send.cpp ###

/\*\* 示例: 向消息队列中发送消息, prio需要从命令行参数中读取 \*\*/

#include <stdio.h>
    #include <stdlib.h>
    #include <mqueue.h>
    
    struct Student  
    {  
        char name[36];  
        int age;  
    };
    
    int main(int argc,char **argv)  
    {  
        if (argc != 2)
        {
            printf("./send <prio>\n");
            return -1;
        }
      
        mqd_t mqid = mq_open("/test", O_WRONLY|O_CREAT, 0666, NULL);  
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
      
        struct Student stu = {"xiaofang", 23};  
        unsigned prio = atoi(argv[1]);
        if (mq_send(mqid, (const char *)&stu, sizeof(stu), prio) == -1)
        {
            printf("mq_send error!\n");
            return -1;
        }
        
        mq_close(mqid);
        return 0;
    }

### 2.5.2 mqueue\_rec.cpp ###

/\*\* 示例: 从消息队列中获取消息 \*\*/

#include <stdio.h>
    #include <stdlib.h>
    #include <mqueue.h>
    
    struct Student  
    {  
        char name[36];  
        int age;  
    };
     
    int main(int argc,char **argv)  
    {  
        mqd_t mqid = mq_open("/test", O_RDONLY);  
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
      
        struct Student buf;  
        int nrcv;  
        unsigned prio;  
        struct mq_attr attr;  
        if (mq_getattr(mqid, &attr) == -1)
        {
            printf("mq_getattr error!\n");
            return -1;
        }
      
        if ((nrcv = mq_receive(mqid, (char *)&buf, attr.mq_msgsize, &prio)) == -1)
        {
            printf("mq_receive error!\n");
            return -1;
        }
      
        printf("receive %d  bytes, priority: %d, name:%s, age:%d\n", nrcv,  prio, buf.name, buf.age);
      
        mq_close(mqid);  
        return 0;  
    }

## 2.6 建立/删除消息到达通知事件 ##

大家在从消息队列接收消息的时候会发现，当消息队列为空的时候，mq\_receive会阻塞，直到有人给队列发送了消息才能返回并继续执行。在很多应用场景下，这种同步处理的方式会给程序本身带来性能瓶颈。为此，POSI消息队列使用mq\_notify为处理过程增加了一个异步通知机制。使用这个机制，我们就可以让队列在由空变成不空的时候触发一个异步事件，通知调用进程，以便让进程可以在队列为空的时候不用阻塞等待。这个方法的原型为：

#include <mqueue.h>
    int mq_notify(mqd_t mqdes, const struct sigevent *sevp);

返回： 成功时为0，出错时为－1  
功能： 给指定队列建立或删除异步事件通知  
sigev\_notify代表通知的方式: 一般常用两种取值:SIGEV\_SIGNAL, 以信号方式通知; SIGEV\_THREAD, 以线程方式通知

如果以信号方式通知: 则需要设定一下两个参数:

sigev\_signo: 信号的代码

sigev\_value: 信号的附加数据(实时信号)

如果以线程方式通知: 则需要设定以下两个参数:

sigev\_notify\_function

sigev\_notify\_attributes

union sigval
    {
        int sival_int;      /* Integer value */
        void *sival_ptr;    /* pointer value */
    };
     
    struct sigevent
    {
        int     sigev_notify;   /* SIGEV_{ NONE, SIGNAL, THREAD} */
        int     sigev_signo;    /* signal number if SIGEV_SIGNAL */
        union sigval sigev_value;   /* passed to signal handler or thread */
        void    (*sigev_notify_function)(union sigval);
        pthread_attr_t *sigev_notify_attribute;
    };

参数sevp:

NULL: 表示撤销已注册通知;

非空: 表示当消息到达且消息队列当前为空, 那么将得到通知;

通知方式:

(1) 产生一个信号, 需要自己绑定

(2) 创建一个线程, 执行指定的函数

注意: 这种注册的方式只是在消息队列从空到非空时才产生消息通知事件, 而且这种注册方式是一次性的!

\*\* Posix IPC所特有的功能, System V没有 \*\*/

### 2.6.1 mqueue\_notify1.cpp ###

/\*\*  
示例: 将下面程序多运行几遍, 尤其是当消息队列"从空->非空", 多次"从空->非空", 当消息队列不空时运行该程序时, 观察该程序的状态;   
\*\*/

#include <stdio.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <signal.h>
    #include <unistd.h>
    
    struct Student  
    {  
        char name[36];  
        int age;  
    };
    
    mqd_t mqid;  
    long size;  
    void sigHandlerForUSR1(int signo)  
    {  
        // 将数据的读取转移到对信号SIGUSR1的响应函数中来  
        struct Student buf;  
        int nrcv;  
        unsigned prio;  
        if ((nrcv = mq_receive(mqid, (char *)&buf, size, &prio)) == -1)
        {
            printf("mq_receive error!\n");
            return;
        }
      
        printf("receive %d bytes, priority: %d, name: %s, age: %d", nrcv, prio, buf.name, buf.age);  
    }  
      
    int main(int argc,char **argv)  
    {  
        // 安装信号响应函数  
        if (signal(SIGUSR1, sigHandlerForUSR1) == SIG_ERR)
        {
            printf("signal error!\n");
            return -1;
        }
      
        mqid = mq_open("/test", O_RDONLY);  
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
      
        // 获取消息的最大长度  
        struct mq_attr attr;  
        if (mq_getattr(mqid, &attr) == -1)
        {
            printf("mq_getattr error!\n");
            return -1;
        }
        
        size = attr.mq_msgsize;  
      
        // 注册消息到达通知事件  
        struct sigevent event;  
        event.sigev_notify = SIGEV_SIGNAL;  // 指定以信号方式通知  
        event.sigev_signo = SIGUSR1;        // 指定以SIGUSR1通知  
        if (mq_notify(mqid, &event) == -1)
        {
            printf("mq_notify error");
            return -1;
        }
      
        //死循环, 等待信号到来  
        while (true)
        {
            printf("x\n");
            sleep(1);
        }
      
        mq_close(mqid);
        
        return 0;  
    }

编译这个程序并执行：

[root@localhost mqueue]# ./mqueue_notify1
    x
    x
    x
    x
    x
    x

会一直打印x，等着队列变为非空，我们此时在别的终端给队列发送一个消息：

[root@localhost mqueue]# ./mqueue_send 123

进程接收到信号，并且现实消息相关内容：

x
    x
    receive 40 bytes, priority: 123, name: xiaofang, age: 23x
    x
    x

再发一个试试：

[root@localhost mqueue]# ./mqueue_send 123

显示：

x
    x
    x
    x

这是因为只注册了一次, 所以后面就失效了。

### 2.6.2 mqueue\_notify2.cpp ###

/\*\*  
示例:多次注册notify, 这样就能过多次接收消息, 但是还是不能从队列非空的时候进行接收, 将程序改造如下:   
\*\*/

#include <stdio.h>
    #include <stdlib.h>
    #include <mqueue.h>
    #include <signal.h>
    #include <unistd.h>
     
    struct Student  
    {  
        char name[36];  
        int age;  
    };
     
    mqd_t mqid;  
    long size;  
    struct sigevent event;
     
    void sigHandlerForUSR1(int signo)  
    {  
        // 注意: 是在消息被读走之前进行注册,  
        // 不然该程序就感应不到消息队列"从空->非空"的一个过程变化了  
        if (mq_notify(mqid, &event) == -1)
        {
            printf("mq_notify error!\n");
            return;
        }
      
        //将数据的读取转移到对信号SIGUSR1的响应函数中来  
        struct Student buf;  
        int nrcv;  
        unsigned prio;  
        if ((nrcv = mq_receive(mqid, (char *)&buf, size, &prio)) == -1)
        {
            printf("mq_receive error!\n");
            return;
        }
      
        printf("receive %d bytes, priority: %d, name: %s, age: %d", nrcv, prio, buf.name, buf.age);
    }  
      
    int main(int argc,char **argv)  
    {  
        // 安装信号响应函数  
        if (signal(SIGUSR1, sigHandlerForUSR1) == SIG_ERR)
        {
            printf("signal error!\n");
            return -1;
        }
      
        mqid = mq_open("/test", O_RDONLY);  
        if (mqid == -1)
        {
            printf("mq_open error!\n");
            return -1;
        }
      
        // 获取消息的最大长度  
        struct mq_attr attr;  
        if (mq_getattr(mqid, &attr) == -1)
        {
            printf("mq_getattr error!\n");
            return -1;
        }
        size = attr.mq_msgsize;  
      
        // 注册消息到达通知事件  
        event.sigev_notify = SIGEV_SIGNAL;  //指定以信号方式通知  
        event.sigev_signo = SIGUSR1;        //指定以SIGUSR1通知  
        if (mq_notify(mqid, &event) == -1)
        {
            printf("mq_notify error!\n");
            return -1;
        }
      
        //死循环, 等待信号到来  
        while (true)
        {
            printf("x\n");
            sleep(1);
        }
      
        mq_close(mqid);  
        return 0;  
    }

mq\_notify 注意点总结:

(1) 任何时刻只能有一个进程可以被注册为接收某个给定队列的通知;

(2) 当有一个消息到达某个先前为空的队列, 而且已有一个进程被注册为接收该队列的通知时, 只有没有任何线程阻塞在该队列的mq\_receive调用的前提下, 通知才会发出;

(3) 当通知被发送给它的注册进程时, 该进程的注册被撤销. 进程必须再次调用mq\_notify以重新注册(如果需要的话),但是要注意: 重新注册要放在从消息队列读出消息之前而不是之后(如同示例程序);

###       2.6.3 异步信号安全函数 ###

#include <signal.h>
    int sigwait(const sigset_t *set, int *sig);

可以使用sigwait函数代替信号处理程序的信号通知，将信号阻塞到某个函数中，仅仅等待该信号的递交。采用sigwait实现上面的程序如下：

#include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <mqueue.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <signal.h>
     
    int main(int argc,char *argv[])
    {
        mqd_t       mqd;
        int         signo;
        void        *buff;
        ssize_t     n;
        sigset_t    newmask;
        struct mq_attr  attr;
        struct sigevent sigev;
        if(argc != 2)
        {
            printf("usage :mqnotify <name>\n");
            exit(0);
        }
        
        mqd = mq_open(argv[1],O_RDONLY);
        mq_getattr(mqd,&attr);
        buff = malloc(attr.mq_msgsize);
        sigemptyset(&newmask);
        sigaddset(&newmask,SIGUSR1);
        sigprocmask(SIG_BLOCK,&newmask,NULL);
        
        sigev.sigev_notify = SIGEV_SIGNAL;
        sigev.sigev_signo = SIGUSR1;
        if(mq_notify(mqd,&sigev) == -1)
        {
            perror("mq_notify error!\n");
            exit(-1);
        }
        for(; ;)
        {
           sigwait(&newmask,&signo); //阻塞并等待该信号
           if(signo == SIGUSR1)
           {
                mq_notify(mqd,&sigev);
                while((n = mq_receive(mqd,(char*)buff,attr.mq_msgsize,NULL))>=0)
                {
                    printf("read %ld bytes\n",(long) n);
                }
                if(errno != EAGAIN)
                {
                    perror("mq_receive error!\n");
                    exit(-1);
                }
           }
        }
        exit(0);
    }

启动线程处理消息通知，程序如下：

#include <stdio.h>
    #include <stdlib.h>
    #include <unistd.h>
    #include <mqueue.h>
    #include <fcntl.h>
    #include <errno.h>
    #include <signal.h>
    #include <pthread.h>
     
    mqd_t       mqd;
    struct mq_attr  attr;
    struct sigevent sigev;
    static void notify_thread(union sigval);
     
    int main(int argc,char *argv[])
    {
     
        if(argc != 2)
        {
            printf("usage :mqnotify <name>\n");
            exit(0);
        }
        mqd = mq_open(argv[1],O_RDONLY | O_NONBLOCK);
        mq_getattr(mqd,&attr);
     
        sigev.sigev_notify = SIGEV_THREAD;
        sigev.sigev_value.sival_ptr = NULL;
        sigev.sigev_notify_function = notify_thread;
        sigev.sigev_notify_attributes = NULL;
     
        if(mq_notify(mqd,&sigev) == -1)
        {
            perror("mq_notify error");
            exit(-1);
        }
        for(; ;)
        {
            pause();
        }
        exit(0);
    }
    
    static void notify_thread(union sigval arg)
    {
        ssize_t     n;
        void        *buff;
        printf("notify_thread started\n");
        buff = malloc(attr.mq_msgsize);
        mq_notify(mqd,&sigev);
        while((n = mq_receive(mqd, (char*)buff,attr.mq_msgsize,NULL))>=0)
        {
            printf("read %ld bytes\n",(long) n);
        }
        
        if(errno != EAGAIN)
        {
            perror("mq_receive error!\n");
            exit(-1);
        }
        free(buff);
        pthread_exit(NULL);
    }

# 3 注意 #

## 3.1 查看已经成功创建的Posix消息队列 ##

\#其存在与一个虚拟文件系统中, 需要将其挂载到系统中才能查看

Mounting the message queue filesystem On Linux, message queues are created in a virtual filesystem.

(Other implementations may also  provide such a feature, but the details are likely to differ.)  This

file system can be mounted (by the superuser, 注意是使用root用户才能成功) using the following commands:

mkdir /dev/mqueue
    
    mount -t mqueue none /dev/mqueue

还可以使用cat查看该消息队列的状态, rm删除:

cat /dev/mqueue/abc 
    
    rm abc

还可umount该文件系统

umount /dev/mqueue

## 3.2 查看具体参数限制 ##

在所有可以显示的属性中，O\_NONBLOCK是mq\_setattr唯一可以更改的参数设置，其他参数对于这个方法都是只读的，不能修改。系统提供了其他手段可以对这些限制进行修改：

(1) /proc/sys/fs/mqueue/msg\_default：在mq\_open的attr参数设置为NULL的时候，这个文件中的数字限定了mq\_maxmsg的值，就是队列的消息个数限制。默认为10个，当消息数达到上限之后，再使用mq\_send发送消息会阻塞。

[root@localhost mqueue]# cat /proc/sys/fs/mqueue/msg_default
    10

(2) /proc/sys/fs/mqueue/msg\_max：可以通过mq\_open的attr参数设定的mq\_maxmsg的数字上限。这个值默认也是10。

[root@localhost mqueue]# cat /proc/sys/fs/mqueue/msg_max
    10

(3) /proc/sys/fs/mqueue/msgsize\_default：在mq\_open的attr参数设置为NULL的时候，这个文件中的数字限定了mq\_msgsize的值，就是队列的字节数数限制。

[root@localhost mqueue]# cat /proc/sys/fs/mqueue/msgsize_default
    8192

(4) /proc/sys/fs/mqueue/msgsize\_max：可以通过mq\_open的attr参数设定的mq\_msgsize的数字上限。

[root@localhost mqueue]# cat /proc/sys/fs/mqueue/msgsize_max
    8192

(5) /proc/sys/fs/mqueue/queues\_max：系统可以创建的消息队列个数上限。

[root@localhost mqueue]# cat  /proc/sys/fs/mqueue/queues_max
    256

**本文转自：**

**[https://blog.csdn.net/NK\_test/article/details/50286309][https_blog.csdn.net_NK_test_article_details_50286309]**

**[https://blog.csdn.net/renwotao2009/article/details/52710206][https_blog.csdn.net_renwotao2009_article_details_52710206]**

[https_blog.csdn.net_NK_test_article_details_50286309]: https://blog.csdn.net/NK_test/article/details/50286309
[https_blog.csdn.net_renwotao2009_article_details_52710206]: https://blog.csdn.net/renwotao2009/article/details/52710206