Unix IPC之Posix消息队列(3)

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 */
};

/* Receive the oldest from highest priority messages in message queue
   MQDES.  */
ssize_t mq_receive (mqd_t __mqdes, char *__msg_ptr, size_t __msg_len,
                    unsigned int *__msg_prio);

/* Add message pointed by MSG_PTR to message queue MQDES.  */
int mq_send (mqd_t __mqdes, __const char *__msg_ptr, size_t __msg_len,
             unsigned int __msg_prio);

程序1：向消息队列中发送消息。可指定消息大小及优先级。

// mqsend.c

#include    "unpipc.h"

int
main(int argc, char **argv)
{
    mqd_t   mqd;
    void    *ptr;
    size_t  len;
    unsigned int  prio;

    if (argc != 4)
        err_quit("usage: mqsend <name> <#bytes> <priority>");
    len = atoi(argv[2]);
    prio = atoi(argv[3]);

    mqd = Mq_open(argv[1], O_WRONLY);

    ptr = Calloc(len, sizeof(char)); // 使用calloc函数分配一个长度为len的缓冲区，默认被初始化为0
    Mq_send(mqd, ptr, len, prio); // 写入消息队列

    exit(0);
}

程序2：从消息队列读出消息。

// mqreveive.c

#include    "unpipc.h"

int
main(int argc, char **argv)
{
    int     c, flags;
    mqd_t   mqd;
    ssize_t n;
    unsigned int prio;
    void    *buff;
    struct mq_attr  attr;

    flags = O_RDONLY;
    while ( (c = Getopt(argc, argv, "n")) != -1)
    {
        switch (c)
        {
            case 'n': // 命令行中带-n表示以非阻塞模式
                flags |= O_NONBLOCK;
                break;
        }
    }
    if (optind != argc - 1)
        err_quit("usage: mqreceive [ -n ] <name>");

    mqd = Mq_open(argv[optind], flags);
    Mq_getattr(mqd, &attr);
    // 这里开辟的缓冲区需要>=消息的最大长度，以便能够容纳接收到的消息
    buff = Malloc(attr.mq_msgsize); // 根据最大消息的大小开辟缓冲区以接收消息

    n = Mq_receive(mqd, buff, attr.mq_msgsize, &prio); // 这里的len长度需要>=消息的最大长度，否则mq_receive会立即返回EMSGSIZE错误
    printf("read %ld bytes, priority = %u
", (long) n, prio);

    exit(0);
}

运行结果：

[dell@localhost pxmsg]$ ls -l /tmp/mqueue/
总用量 0
[dell@localhost pxmsg]$ ./mqcreate1 /msgqueue
[dell@localhost pxmsg]$ ls -l /tmp/mqueue/
总用量 0

[dell@localhost pxmsg]$ sudo mount -t mqueue none /tmp/mqueue
[sudo] password for dell: 
[dell@localhost pxmsg]$ ls -l /tmp/mqueue/
总用量 0
-rw-r--r--. 1 dell dell 80 8月  13 20:52 msgqueue

[dell@localhost pxmsg]$ ./mqsend /msgqueue 10 1
[dell@localhost pxmsg]$ ./mqsend /msgqueue 20 2
[dell@localhost pxmsg]$ ./mqsend /msgqueue 30 3
[dell@localhost pxmsg]$ ls -l /tmp/mqueue/
总用量 0
-rw-r--r--. 1 dell dell 80 8月  13 20:55 msgqueue

[dell@localhost pxmsg]$ ./mqreceive /msgqueue
read 30 bytes, priority = 3
[dell@localhost pxmsg]$ ./mqreceive /msgqueue
read 20 bytes, priority = 2
[dell@localhost pxmsg]$ ./mqreceive /msgqueue
read 10 bytes, priority = 1
[dell@localhost pxmsg]$ ./mqreceive /msgqueue
^C
// 这里默认是阻塞方式，当消息队列为空时将会被阻塞，所以下面我们使用非阻塞方式。
[dell@localhost pxmsg]$ ./mqreceive -n /msgqueue
mq_receive error: Resource temporarily unavailable
[dell@localhost pxmsg]$