Unix/Linux编程：System V 消息队列

「爱情、让人受尽委屈。」 2022-10-18 00:51 130阅读 0赞

消息队列允许进程以消息的形式交换数据。尽管消息对象在某些方面与管道和FIFO类似，但是它们之间扔存在显著差别

*  用来引用消息队列的句柄是一个由msgget()调用返回的标识符。这些标识符与Unix系统上的大多数其他形式的IO所使用的文件描述符是不同的
 *  通过消息对象进行的通信是面向消息的，即读者接收到由写者写入的整条消息。读取一条消息的一部分而让剩余部分遗留在队列中或者一次读取多条消息都是不可能的。这一点和管道不同，管道提供的是一个无法进程区分的字节流(即使用管道时读者一次可以读取任意数量的字节数，而不管写者写入的数据块大小)
 *  除了包含数据之外，每条消息还有一个用整数表示的类型。从消息对象中读取消息既可以按照先入先出的顺序，也可以根据类型来读取数据

# 创建或打开一个消息队列 #

msgget()系统调用创建一个新消息队列或者取得一个既有队列的标识符：

NAME
           msgget - get a System V message queue identifier
    
    SYNOPSIS
           #include <sys/types.h>
           #include <sys/ipc.h>
           #include <sys/msg.h>
    
           int msgget(key_t key, int msgflg);

key 参数是使用[这里][Link 1]描述的方法之一生成的一个键（即通常是值 IPC\_PRIVATE 或ftok()返回的一个键）

msgflg 参数是一个指定施加于新消息队列之上的权限或检查一个既有队列的权限的位掩码。此外，在 msgflg 参数中还可以将下列标记中的零个或多个标记取 OR（|）以控制 msgget()的操作

*  IPC\_CREAT ：如果没有与指定的 key 对应的消息队列，那么就创建一个新队列
 *  IPC\_EXCL ：如果同时还指定了 IPC\_CREAT 并且与指定的 key 对应的队列已经存在，那么调用就会失败并返回 EEXIST 错误。

msgget()系统调用首先会在所有既有消息队列中搜索与指定的键对应的队列。如果找到了一个匹配的队列，那么就会返回该对象的标识符（除非在 msgflg 中同时指定了 IPC\_CREAT和 IPC\_EXCL，那样的话就返回一个错误）。如果没有找到匹配的队列并且在 msgflg 中指定了IPC\_CREAT，那么就会创建一个新队列并返回该队列的标识符。

看个例子：

#define _GNU_SOURCE
    #include <string.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <zconf.h>
    #include <sys/types.h>
    #include <sys/ipc.h>
    #include <sys/msg.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <cstdarg>
    
    
    static void usageError(const char *progName, const char *msg)
    {
        
        if (msg != NULL)
            fprintf(stderr, "%s", msg);
        fprintf(stderr, "Usage: %s [-cx] {-f pathname | -k key | -p} "
                        "[octal-perms]\n", progName);
        fprintf(stderr, "    -c           Use IPC_CREAT flag\n");
        fprintf(stderr, "    -x           Use IPC_EXCL flag\n");
        fprintf(stderr, "    -f pathname  Generate key using ftok()\n");
        fprintf(stderr, "    -k key       Use 'key' as key\n");
        fprintf(stderr, "    -p           Use IPC_PRIVATE key\n");
        exit(EXIT_FAILURE);
    }
    void
    cmdLineErr(const char *format, ...)
    {
        
        va_list argList;
    
        fflush(stdout);           /* Flush any pending stdout */
    
        fprintf(stderr, "Command-line usage error: ");
        va_start(argList, format);
        vfprintf(stderr, format, argList);
        va_end(argList);
    
        fflush(stderr);           /* In case stderr is not line-buffered */
        exit(EXIT_FAILURE);
    }
    int main(int argc, char *argv[])
    {
        
        int numKeyFlags;            /* Counts -f, -k, and -p options */
        int flags, msqid, opt;
        unsigned int perms;
        long lkey;
        key_t key;
    
        /* Parse command-line options and arguments */
    
        numKeyFlags = 0;
        flags = 0;
    
        while ((opt = getopt(argc, argv, "cf:k:px")) != -1) {
        
            switch (opt) {
        
                case 'c':
                    flags |= IPC_CREAT;
                    break;
    
                case 'f':               /* -f pathname */
                    key = ftok(optarg, 1);
                    if (key == -1){
        
                        perror("ftok");
                        exit(EXIT_FAILURE);
                    }
    
                    numKeyFlags++;
                    break;
    
                case 'k':               /* -k key (octal, decimal or hexadecimal) */
                    if (sscanf(optarg, "%li", &lkey) != 1)
                        cmdLineErr("-k option requires a numeric argument\n");
                    key = lkey;
                    numKeyFlags++;
                    break;
    
                case 'p':
                    key = IPC_PRIVATE;
                    numKeyFlags++;
                    break;
    
                case 'x':
                    flags |= IPC_EXCL;
                    break;
    
                default:
                    usageError(argv[0], "Bad option\n");
            }
        }
    
        if (numKeyFlags != 1)
            usageError(argv[0], "Exactly one of the options -f, -k, "
                                "or -p must be supplied\n");
    
        perms = (optind == argc) ? (S_IRUSR | S_IWUSR) :atoi(argv[optind]);  //"octal-perms"
    
        msqid = msgget(key, flags | perms);
        if (msqid == -1){
        
            perror("msgget");
            exit(EXIT_FAILURE);
        }
    
    
        printf("%d\n", msqid);
        exit(EXIT_SUCCESS);
    }

# 交换消息 #

`msgsnd()`和`msgrcv()`系统调用执行消息队列上的IO

*  msgsnd()系统调用向消息队列写入一条消息。

NAME
           msgrcv, msgsnd - System V message queue operations
    
    SYNOPSIS
           #include <sys/types.h>
           #include <sys/ipc.h>
           #include <sys/msg.h>
    
           int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
    
           ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,
                          int msgflg);
    
    DESCRIPTION
    		参数msqid表示队列标识符。
    
    		参数msgp是一个由程序员定义的结构的指针，该结构用于存放被发送或接收的消息。这个结构的常规形式如下。
    
    	    struct msgbuf {
        
                   long mtype;       /* message type, must be > 0 */
                   char mtext[];    /* message data */
               };
    RETURN VALUE
           失败时，两个函数都返回 -1，其中 errno 表示错误，否则 msgsnd() 返回 0，而 msgrcv() 返回实际复
           制到 mtext 数组中的字节数。

使用 msgsnd()发送消息时：

*  必须要将消息结构中的 mtype 字段的值设为一个大于 0 的值。
 *  并将所需传递的信息复制到程序员定义的 mtext字段中。
 *  msgsz 参数指定了 mtext 字段中包含的字节数。
 *  参数 msgflg 是一组标记的位掩码，用于控制 msgsnd()的操作，目前只定义了一个这样的标记
    
     *  IPC\_NOWAIT ：
        
         *  执行一个非阻塞的发送操作。
         *  通常，当消息队列满时，msgsnd()会阻塞直到队列中有足够的空间来存放这条消息。但如果指定了这个标记，那么 msgsnd()就会立即返回 EAGAIN 错误。

*  当msgsnd()调用因队列满而发生阻塞时可能会被信号处理器中断。当发生这种情况时，msgsnd()总是会返回EINTR错误（msgsnd()系统调用永远不会自动重启，不管在建立信号处理器时是否设置了 SA\_RESTART 标记）
 *  向消息队列中写消息要求具备在该队列上的写权限。

msgrcv()系统调用从消息队列中读取（以及删除）一条消息并将其内容复制进 msgp 指向的缓冲区中。

*  msgp 缓冲区中 mtext 字段的最大可用空间是通过 msgsz 参数来指定的。如果队列中待删除的消息体的大小超过了 msgsz字节，那么就不会从队列中删除消息，并且 msgrcv()会返回错误 E2BIG。（这是默认行为，可以使用 MSG\_NOERROR 标记来改变这种行为）
 *  读取消息的顺序无需与消息被发送的一致。可以根据 mtype 字段的值来选择消息，而这个选择过程是由 msgtyp 参数来控制的，具体如下所述。
    
     *  如果 msgtyp 等于 0，那么会删除队列中的第一条消息并将其返回给调用进程。
     *  如果 msgtyp 大于 0，那么会将队列中第一条 mtype 等于 msgtyp 的消息删除并将其返回给调用进程。通过指定不同的 msgtyp 值，多个进程能够从同一个消息队列中读取消息而不会出现竞争读取同一条消息的情况。比较有用的一项技术是让各个进程选取与自己的进程 ID 匹配的消息
     *  如果 msgtyp 小于 0，那么就会将等待消息当成优先队列来处理。队列中 mtype 最小并且其值小于或等于 msgtyp 的绝对值的第一条消息会被删除并返回给调用进程。

下面通过一个例子将讲解 msgtyp 小于 0 时的情况。假设一个消息队列包含了下图显示的一组消息，接着执行一系列的 msgrcv()调用，其形式如下。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]

msgrcv(id, &msg, maxmsgsz, -300, 0);

这些 msgrcv()调用会按照 2（类型为 100）、5（类型为 100）、3（类型为 200）、1（类型为300）的顺序读取消息。后续的调用会阻塞，因为剩余的消息的类型（400）超过了 300

*  msgflg 参数是一个位掩码，它的值通过将下列标记中的零个或多个取 OR 来确定
    
     *  IPC\_NOWAIT
        
         *  执行一个非阻塞接收。
         *  通常如果队列中没有匹配 msgtyp 的消息，那么 msgrcv()会阻塞直到队列中存在匹配的消息为止。指定 IPC\_NOWAIT 标记会导致 msgrcv()立即返回 ENOMSG错误
     *  MSG\_EXCEPT
        
         *  只有当 msgtyp 大于 0 时这个标记才会起作用，它会强制对常规操作进行补足，即将队列中第一条 mtype 不等于 msgtyp 的消息删除并将其返回给调用者。
         *  这个标记是 Linux 特有的，只有当定义了\_GNU\_SOURCE 之后才会在<sys/msg.h>中提供这个标记
         *  在上图中给出的消息队列上执行一系列形式为 msgrcv(id, &msg, maxmsgsz, 100, MSG\_EXCEPT)的调用将会按照1、3、4 顺序读取消息，之后发生阻塞
     *  MSG\_NOERROR
        
         *  在默认情况下，当消息的 mtext 字段的大小超过了可用空间时（由 maxmsgsz 参数定义），msgrcv()调用会失败。
         *  如果指定了 MSG\_ NOERROR 标记，那么 msgrcv()将会从队列中删除消息并将其 mtext 字段的大小截短为maxmsgsz 字节，然后将消息返回给调用者。被截去的数据将会丢失。
 *  msgrcv()成功完成之后会返回接收到的消息的 mtext 字段的大小，发生错误时则返回−1。
 *  与 msgsnd()一样，如果被阻塞的 msgrcv()调用被一个信号处理器中断了，那么调用会失败并返回 EINTR 错误，不管在建立信号处理器时是否设置了 SA\_RESTART 标记
 *  从消息队列中读取消息需要具备在队列上的读权限

看个例子：

*  使用 msgsnd()发送一条消息

#define _GNU_SOURCE
    #include <string.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <zconf.h>
    #include <sys/types.h>
    #include <sys/ipc.h>
    #include <sys/msg.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <cstdarg>
    
    
    #define MAX_MTEXT 1024
    
    struct mbuf {
        
        long mtype;                         /* Message type */
        char mtext[MAX_MTEXT];              /* Message body */
    };
    static void             /* Print (optional) message, then usage description */
    usageError(const char *progName, const char *msg)
    {
        
        if (msg != NULL)
            fprintf(stderr, "%s", msg);
        fprintf(stderr, "Usage: %s [-n] msqid msg-type [msg-text]\n", progName);
        fprintf(stderr, "    -n       Use IPC_NOWAIT flag\n");
        exit(EXIT_FAILURE);
    }
    
    void
    cmdLineErr(const char *format, ...)
    {
        
        va_list argList;
    
        fflush(stdout);           /* Flush any pending stdout */
    
        fprintf(stderr, "Command-line usage error: ");
        va_start(argList, format);
        vfprintf(stderr, format, argList);
        va_end(argList);
    
        fflush(stderr);           /* In case stderr is not line-buffered */
        exit(EXIT_FAILURE);
    }
    int main(int argc, char *argv[])
    {
        
        int msqid, flags, msgLen;
        struct mbuf msg;                    /* Message buffer for msgsnd() */
        int opt;                            /* Option character from getopt() */
    
        /* Parse command-line options and arguments */
        flags = 0;
        while ((opt = getopt(argc, argv, "n")) != -1) {
        
            if (opt == 'n')
                flags |= IPC_NOWAIT;
            else
                usageError(argv[0], NULL);
        }
    
        if (argc < optind + 2 || argc > optind + 3)
            usageError(argv[0], "Wrong number of arguments\n");
    
        msqid = atoi(argv[optind]);
        msg.mtype = atoi(argv[optind + 1]);
    
        if (argc > optind + 2) {
                    /* 'msg-text' was supplied */
            msgLen = strlen(argv[optind + 2]) + 1;
            if (msgLen > MAX_MTEXT)
                cmdLineErr("msg-text too long (max: %d characters)\n", MAX_MTEXT);
    
            memcpy(msg.mtext, argv[optind + 2], msgLen);
    
        } else {
                                    /* No 'msg-text' ==> zero-length msg */
            msgLen = 0;
        }
    
        /* Send message */
    
        if (msgsnd(msqid, &msg, msgLen, flags) == -1){
        
            perror("msgsnd");
            exit(EXIT_FAILURE);
        }
            
        exit(EXIT_SUCCESS);
    }

*  使用 msgrcv()读取一条消息

#define _GNU_SOURCE
    #include <string.h>
    #include <stdio.h>
    #include <stdlib.h>
    #include <zconf.h>
    #include <sys/types.h>
    #include <sys/ipc.h>
    #include <sys/msg.h>
    #include <sys/stat.h>
    #include <errno.h>
    #include <cstdarg>
    
    
    void
    cmdLineErr(const char *format, ...)
    {
        
        va_list argList;
    
        fflush(stdout);           /* Flush any pending stdout */
    
        fprintf(stderr, "Command-line usage error: ");
        va_start(argList, format);
        vfprintf(stderr, format, argList);
        va_end(argList);
    
        fflush(stderr);           /* In case stderr is not line-buffered */
        exit(EXIT_FAILURE);
    }
    #define MAX_MTEXT 1024
    
    struct mbuf {
        
        long mtype;                 /* Message type */
        char mtext[MAX_MTEXT];      /* Message body */
    };
    static void
    usageError(const char *progName, const char *msg)
    {
        
        if (msg != NULL)
            fprintf(stderr, "%s", msg);
        fprintf(stderr, "Usage: %s [options] msqid [max-bytes]\n", progName);
        fprintf(stderr, "Permitted options are:\n");
        fprintf(stderr, "    -e       Use MSG_NOERROR flag\n");
        fprintf(stderr, "    -t type  Select message of given type\n");
        fprintf(stderr, "    -n       Use IPC_NOWAIT flag\n");
    #ifdef MSG_EXCEPT
        fprintf(stderr, "    -x       Use MSG_EXCEPT flag\n");
    #endif
        exit(EXIT_FAILURE);
    }
    int
    main(int argc, char *argv[])
    {
        
        int msqid, flags, type;
        ssize_t msgLen;
        size_t maxBytes;
        struct mbuf msg;            /* Message buffer for msgrcv() */
        int opt;                    /* Option character from getopt() */
    
        /* Parse command-line options and arguments */
    
        flags = 0;
        type = 0;
        while ((opt = getopt(argc, argv, "ent:x")) != -1) {
        
            switch (opt) {
        
                case 'e':       flags |= MSG_NOERROR;   break;
                case 'n':       flags |= IPC_NOWAIT;    break;
                case 't':       type = atoi(optarg);    break;
    #ifdef MSG_EXCEPT
                case 'x':       flags |= MSG_EXCEPT;    break;
    #endif
                default:        usageError(argv[0], NULL);
            }
        }
    
        if (argc < optind + 1 || argc > optind + 2)
            usageError(argv[0], "Wrong number of arguments\n");
    
        msqid = atoi(argv[optind]);
        maxBytes = (argc > optind + 1) ? atoi(argv[optind + 1]): MAX_MTEXT;
    
        /* Get message and display on stdout */
    
        msgLen = msgrcv(msqid, &msg, maxBytes, type, flags);
        if (msgLen == -1){
        
            perror("msgrcv");
            exit(EXIT_FAILURE);
        }
    
    
        printf("Received: type=%ld; length=%ld", msg.mtype, (long) msgLen);
        if (msgLen > 0)
            printf("; body=%s", msg.mtext);
        printf("\n");
    
        exit(EXIT_SUCCESS);
    }

*  使用

（1） 首先使用 IPC\_PRIVATE 键创建了一个消息队列，然后向队列中写入了三条不同类型的消息。  
![在这里插入图片描述][20210604102110387.png]  
（2）从队列中读取类型小于或等于 20 的消息  
![在这里插入图片描述][2021060410213069.png]  
（3）上面最后一条命令会阻塞，因为队列中已经没有类型小于或等于 20 的消息了。因此需要输入 Control-C 来终止这个命令，然后执行一个从队列中读取任意类型的消息的命令。  
![在这里插入图片描述][20210604102150566.png]

# 消息队列控制操作 #

msgctl()系统调用在标识符为 msqid 的消息队列上执行控制操作。

NAME
           msgctl - System V message control operations
    
    SYNOPSIS
           #include <sys/types.h>
           #include <sys/ipc.h>
           #include <sys/msg.h>
    
           int msgctl(int msqid, int cmd, struct msqid_ds *buf);
    
    DESCRIPTION
           msgctl() performs the control operation specified by cmd on the System V message queue with identifier msqid.
    
           The msqid_ds data structure is defined in <sys/msg.h> as follows:
    
               struct msqid_ds {
        
                   struct ipc_perm msg_perm;     /* Ownership and permissions */
                   time_t          msg_stime;    /* Time of last msgsnd(2) */
                   time_t          msg_rtime;    /* Time of last msgrcv(2) */
                   time_t          msg_ctime;    /* Time of last change */
                   unsigned long   __msg_cbytes; /* Current number of bytes in
                                                    queue (nonstandard) */
                   msgqnum_t       msg_qnum;     /* Current number of messages
                                                    in queue */
                   msglen_t        msg_qbytes;   /* Maximum number of bytes
                                                    allowed in queue */
                   pid_t           msg_lspid;    /* PID of last msgsnd(2) */
                   pid_t           msg_lrpid;    /* PID of last msgrcv(2) */
               };
    
           The ipc_perm structure is defined as follows (the highlighted fields are settable using IPC_SET):
    
               struct ipc_perm {
        
                   key_t          __key;       /* Key supplied to msgget(2) */
                   uid_t          uid;         /* Effective UID of owner */
                   gid_t          gid;         /* Effective GID of owner */
                   uid_t          cuid;        /* Effective UID of creator */
                   gid_t          cgid;        /* Effective GID of creator */
                   unsigned short mode;        /* Permissions */
                   unsigned short __seq;       /* Sequence number */
               };

cmd 参数指定了在队列上执行的操作，其取值是下列值中的一个

*  IPC\_RMID：立即删除消息队列对象及其关联的 msqid\_ds 数据结构。队列中所有剩余的消息都会丢失，所有被阻塞的读者和写者进程会立即醒来，msgsnd()和 msgrcv()会失败并返回错误 EIDRM。这个操作会忽略传递给 msgctl()的第三个参数。也就是说，第三个参数应为NULL
 *  IPC\_STAT：将与这个消息队列关联的 msqid\_ds 数据结构的副本放到 buf 指向的缓冲区中。
 *  IPC\_SET：使用buf指向的缓冲区提供的值更新与这个消息队列关联的msqid\_ds数据结构中被选中的字段。

# 缺点 #

Unix系统为同一系统上不同进程之间的数据传输提供了很多机制，既包含无分隔符的字节流形式(管道、FIFO、Unix domain流socket)，也包括有分隔符的消息形式(System V 消息队列，POSIX消息队列以及Unix domain数据报socket)

System V消息队列的一个与众不同的特性是它能够为每个消息队列加上一个数字类型。应用程序可以使用这个完成两件事情：读取进程可以根据类型来选择消息或者它们可以采用一种优先队列策略以便优先读取高优先级的消息（即那些消息类型值更低的消息）

但 System V 消息队列也存在几个缺点。

*  消息队列是基于标识符引用的，而不是像大多数其他Unix IO机制那样使用文件描述符，这意味着那些基于文件描述符的IO技术比如select、poll、epoll将无法应用于消息队列上。此外，在程序中编写同时处理消息队列的输入和基于文件描述符的 I/O 机制的代码要比编写只处理文件描述符的代码更加复杂
 *  使用key而不是文件名来标识消息队列会增加额外的程序设计复杂性，同时还需要使用ipcs 和 ipcrm 来替换 ls 和 rm。ftok()函数通常能产生一个唯一的键，但却无法保证一定能产生。使用IPC\_PRIVATE键能够确保产生唯一的队列标识符，但是需要使这个标识符对需要用到它的其他进程可见
 *  消息队列是无连接的，内核不会像对待管道、FIFO以及socket那样维护引用队列的进程数。因此就难以回答下列问题。
    
     *  一个应用程序何时能够安全地删除一个消息队列？
     *  应用程序如何确保不再使用的队列会被删除呢
 *  消息队列的总数、消息的大小以及单个队列的容量都是有限制的。这些限制都是可配置的，但如果一个应用程序超出了这些默认限制的范围，那么在安装应用程序的时候就需要完成一些额外的工作了

由于存在这些限制，**新应用程序应该尽可能的避免使用System V 消息队列**，而应该使用其他形式的 IPC机制，如 POSIX 消息队列、FIFO以及 socket。

# 总结 #

System V 消息队列允许进程通过交换由一个数字类型和一个包含任意数据的消息体构成的消息的形式来进行通信。消息队列的区别于其他机制的特性是消息是有边界的，并且接收者能够根据类型来选择消息，而无需按照先入先出的顺序来读取消息。

之所以得出其他 IPC 机制通常要优于 System V 消息队列的结论是因为几个因素，其中最主要的一个是引用消息队列不会用到文件描述符。这意味着在消息队列上无法使用另一种 I/O模型，特别是同时监控消息队列和文件描述符以查看是否可进行 I/O 将变得复杂。此外，消息队列无连接（即不进行引用计数）这个事实使得应用程序难以知道何时能够安全地删除一个队列

[Link 1]: https://blog.csdn.net/zhizhengguan/article/details/117527276
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poaXpoZW5nZ3Vhbg_size_16_color_FFFFFF_t_70]: /images/20221014/995b96a7039740618c36980bf082f7c6.png
[20210604102110387.png]: /images/20221014/13aa82aae97449fb9189e7cb0588fbfe.png
[2021060410213069.png]: /images/20221014/ca06bebb8b72451a8da5073d8bf8ea73.png
[20210604102150566.png]: /images/20221014/773a5fd3f5ac4348ae9a56cedbde8763.png