《unix高级环境编程》进程间通信——消息队列

墨蓝 2022-08-14 00:55 239阅读 0赞

消息队列是消息的链接表，保存在内核，通过消息队列的引用标识符来访问消息，消息队列对每个消息指定了特定的消息类型，接收消息的进程可以请求接收下一条消息，也可以请求接收下一条特定类型的消息。系统内核维护的消息队列的结构如下：

#include <bits/msq.h>
    
    struct msqid_ds
    {
      struct ipc_perm msg_perm;     /* IPC对象的属性信息和访问权限 */
      struct msg *msg_first;        /* 指向消息队列的第一个消息 */
      struct msg *msg_last;         /* 指向消息队列的最后一个消息 */
      time_t msg_stime;             /* time of last msgsnd command */
      time_t msg_rtime;             /* time of last msgrcv command */
      time_t msg_ctime;             /* time of last change */
      unsigned long int msg_cbytes; /* 当前消息队列中消息的总字节数 */
      msgqnum_t msg_qnum;           /* 当前队列中消息的个数 */
      msglen_t msg_qbytes;          /* 队列允许存放的最大字节数 */
      pid_t msg_lspid;              /* pid of last msgsnd() 即最后执行msgsnd函数的进程的进程ID */
      pid_t msg_lrpid;              /* pid of last msgrcv() 即最后执行msgrcv函数的进程的进程ID */
    
    };

其中 ipc\_perm 的结构如下：

struct ipc_perm
    {
        uid_t uid;      /* owner's effective user id */
        gid_t gid;      /* owner's effective group id */
        uid_t cuid;     /* creator's effective user id */
        gid_t cgid;     /* creator's effective user id */
        mode_t mode;    /* access modes */
    };

内核维护的消息队列链表结构形式如下图所示：

![Center][]

消息队列所传递的信息有两部分组成，即消息类型及其所传递数据，一般用一个结构表示，通常消息类型是一个正的长整型数表示，而数据根据需要设定，例如设定一个传送1024字节长度的字符数据的消息结构如下。获取消息队列中的消息时，不一定按照先进先出顺序，也可以按照消息的类型字段进行获取。

struct msgbuf{
        long msgtype;
        char msgtext[1024];
    };

## 消息队列的创建与打开 ##

要是有消息队列，首先必须要创建一个消息队列，msgget 函数能够实现该功能：

/*
     * 函数功能：创建一个新的消息队列或打开一个现有的消息队列；
     * 返回值：若成功则返回消息队列的ID，若出错则返回-1；
     * 函数原型：
     */
    #include <sys/msg.h>
    
    int msgget(key_t key, int flag);
    /*
     * 说明：
     * 参数key是消息队列的键；
     * 参数flag表示调用函数的操作类型，也可用于设置访问权限；
     */

当成功创建一个新消息队列时，msqid\_ds 结构的成员被初始化为如下值：

1.  msg\_perm 结构的 uid 和 cuid 成员被设置成当前进程的有效用户ID，gid 和 cgid 成员被设置成当前进程的有效组ID；
2.  flag中的读写权限位存放在 msg\_perm.mode 中；
3.  msg\_qnum, msg\_lspid, msg\_lrpid, msg\_stime 和 msg\_rtime 被置为0；
4.  msg\_ctime 被设置成当前时间；
5.  msg\_qbytes被设置成系统限制值；

创建一个消息队列的测试程序：

#include "apue.h"
    #include <fcntl.h>
    #include <sys/msg.h>
    
    
    #define  PATH_NAME "./Queue"
    
    int main(void)
    {
        key_t key;
        int fd;
    
        if ((fd = open(PATH_NAME, O_CREAT, 0666)) < 0)
            err_quit("open error");
        close(fd);
        //生成键值
        key = ftok(PATH_NAME, 0);
        int msgID;
    
        if ((msgID = msgget(key, IPC_CREAT | 0666)) == -1)
            err_quit("msgget error");
        printf("key: %x\n", key);
        printf("msgID: %d\n", msgID);
        exit(0);
    
    }

输出结果：

./msg 
    key: 1309d
    msgID: 0

创建消息队列之后可以在终端上输入命令查看消息队列情况：

$ ipcs -q -i 0
    
    Message Queue msqid=0
    uid=1000	gid=1000	cuid=1000	cgid=1000	mode=0666
    cbytes=0	qbytes=16384	qnum=0	lspid=0	lrpid=0
    send_time=Not set                   
    rcv_time=Not set                   
    change_time=Mon Nov 17 14:35:15 2014

## 消息队列的操作 ##

向消息队列中发送消息，我们可以调用 msgsnd 函数实现该功能：

/*
     * 函数功能：向消息队列中发送消息；
     * 返回值：若成功则返回0，若出错则返回-1；
     * 函数原型：
     */
    #include <sys/msg.h>
    
    int msgsnd(int msqid, const void *ptr, size_t nbytes, int flag);
    /*
     * msqid是消息队列的引用标识符；
     * ptr是一个void指针，指向要发送的消息；
     * nbytes表示要发送消息的字节数；
     * flag用于指定消息队列已满时的处理方法，当消息队列为满时，若设置为IPC_NOWAIT，则立刻出错返回EAGAIN；
     * 否则发送消息的进程被阻塞，直到消息队列中空间或消息队列被删除或捕捉到信号时，函数返回；
     */

从消息队列中接收消息，我们可以调用 msgrcv 函数实现该功能：

/*
     * 函数功能：从消息队列中接收消息；
     * 返回值：若成功则返回消息的数据部分的长度，若出错则返回-1；
     * 函数原型：
     */
    #include <sys/msg.h>
    
    ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int flag);
    /*
     * msqid是消息队列的引用标识符；
     * ptr是一个void指针，指向要存放消息数据的缓冲区；
     * nbytes表示要存放消息数据缓冲区的长度；
     *
     * 当返回的消息实际长度大于nbytes时，根据flag的设置进行处理：若设置为MSG_NOERROR，则消息被截短，否则出错返回E2BIG；
     * 若指定的type无效时，若flag设置为IPC_NOWAIT，则立即出错返回，且errno设为ENOMSG，否则接收消息的进程将被阻塞，
     * 直到type有效或者消息队列被删除或者捕捉到信号；
     *
     * type的取值如下：
     * （1）type=0  接收消息队列中的第一条消息；
     * （2）type>0  接收消息队列中类型为type的第一条消息；
     * （3）type<0  接收消息队列中类型值小于或等于type绝对值的所有消息中类型最小的消息中的第一条消息；
     * type为非0，则用于非先进先出顺序读消息；
     */

## 消息队列的控制 ##

对消息队列的具体控制操作可以通过函数 msgctl 来实现：

/*
     * 函数功能：消息队列的控制；
     * 返回值：若成功则返回0，若出错则返回-1；
     * 函数原型：
     */
    #include <sys/msg.h>
    int msgctl(int msqid, int cmd, struct msqid_ds *buf);

根据 cmd 不同的取值有不同的操作，其中 cmd 参数取值如下：

1.  IPC\_STAT：获取消息队列中的 msqid\_ds 结构，并把它保存在 buf 指向的缓冲区；
2.  IPC\_SET：按参数 buf 指向的结构中的值设置该消息队列对应的 msqid\_ds 结构中四个字段—— msg\_perm.uid，msg\_perm.gid，msg\_perm.mode 和 msg\_qbytes。此操作只能由以下两种进程执行：一种是其有效用户 ID 等于 msg\_perm.cuid 或 msg\_perm.uid；另一种是具有超级用户权限的进程。只有超级用户才能增加 msg\_qbytes 的值；
3.  IPC\_RMID：从系统中删除该消息队列以及仍在该队列中的所有数据，该执行立即生效；若还有进程对次消息队列进行操作，则出错返回 EIDRM。此操作只能由以下两种进程执行：一种是其有效用户 ID 等于 msg\_perm.cuid 或 msg\_perm.uid；另一种是具有超级用户权限的进程。

测试程序：

#include "apue.h"
    #include <sys/msg.h>
    #include <fcntl.h>
    
    #define PATH_NAME "./Queue"
    
    key_t MakeKey(const char *pathname);
    
    int main(void)
    {
        int msgid;
        int status;
        key_t key;
        key = MakeKey(PATH_NAME);
    
        char str1[] = "test message: Wellcome.";
        char str2[] = "test message: goodbye.";
        struct msgbuf
        {
            long msgtype;
            char msgtext[MAXLINE];
        }sndmsg, rcvmsg;
    
        if((msgid = msgget(key, IPC_CREAT | 0666)) == -1)
            err_quit("msgget error");
      
            sndmsg.msgtype = 100;
            sprintf(sndmsg.msgtext, str1);
            if(msgsnd(msgid, (struct msgbuf *)&sndmsg, sizeof(str1)+1, 0) == -1)
                err_quit("msgsnd error");
            sndmsg.msgtype = 200;
            sprintf(sndmsg.msgtext, str2);
            if(msgsnd(msgid, (struct msgbuf *)&sndmsg, sizeof(str2)+1, 0) == -1)
                err_quit("msgsnd error");
    
    
        if((status = msgrcv(msgid, (struct msgbuf*)&rcvmsg, 128, 100, IPC_NOWAIT)) == -1)
            err_quit("msgrcv error");
    
        printf("Recevied message:\n%s\n", rcvmsg.msgtext);
    
        if((status = msgrcv(msgid, (struct msgbuf*)&rcvmsg, 128, 200, IPC_NOWAIT)) == -1)
            err_quit("msgrcv error");
    
        printf("Recevied message:\n%s\n", rcvmsg.msgtext);
        msgctl(msgid, IPC_RMID, 0);
        exit(0);
    }
    key_t MakeKey(const char *pathname)
    {
        int fd;
        if((fd = open(pathname, O_CREAT, 0666)) < 0)
            err_quit("open error");
        close(fd);
    
        return ftok(pathname, 0);
    }

输出结果：

$ ./msg
    Recevied message:
    test message: Wellcome.
    Recevied message:
    test message: goodbye.

参考资料：

《UNIX高级环境编程》

[Center]: /images/20220810/e2521657a5454036926fbf9900cff16d.png