进程间通信 posix-消息队列教程

Posix消息队列

消息队列可以认为是一个消息链表，某个进程往一个消息队列中写入消息之前，不需要另外某个进程在该队列上等待消息的达到，这一点与管道和FIFO相反。Posix消息队列与System
V消息队列的区别如下：

1. 对Posix消息队列的读总是返回最高优先级的最早消息，对System V消息队列的读则可以返回任意指定优先级的消息。
2. 当往一个空队列放置一个消息时，Posix消息队列允许产生一个信号或启动一个线程，System V消息队列则不提供类似的机制。

#include    <mqueue.h>
typedef int mqd_t;
mqd_t mq_open(const char *name, int oflag, ... /* mode_t mode, struct mq_attr *attr */);
返回： 成功时为消息队列描述字，出错时为－1。   
功能： 创建一个新的消息队列或打开一个已存在的消息的队列。

#include    <mqueue.h>
int mq_close(mqd_t mqdes);
返回： 成功时为0，出错时为-1。
功能： 关闭已打开的消息队列。

#include    <mqueue.h>
int mq_unlink(const char *name)
返回： 成功时为0，出错时为－1
功能： 从系统中删除消息队列。

#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

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

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

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

\#define MQ\_PRIO\_MAX 32768

#include    <mqueue.h>
int mq_send(mqd_t mqdes, const char *ptr, size_t len, unsigned int prio);
返回：成功时为0，出错为－1
ssize_t mq_receive(mqd_t mqdes, char *ptr, size_t len, unsigned int *priop);
返回：成功时为消息中的字节数，出错为－1

消息队列的限制：

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

#include    <mqueue.h>
int mq_notify(mqd_t mqdes, const struct sigevent *notification);
返回： 成功时为0，出错时为－1
功能： 给指定队列建立或删除异步事件通知

union sigval
{
    int sival_int;      /* Integer value */
    void *sival_ptr;    /* pointer value */
};

struct sigevent
{
    int     sigev_notify;   /* SIGEV_{ NONE, ISGNAL, 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;
}; 

异步信号安全函数

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

Posxi实时信号

信号可划分为两大小组：

1. 其值在SIGRTMIN和SIGRTMAX之间（包括两者在内）的实时信号。
2. 所有其他信号：SIGALRM， SIGINT， SIGKILL等等。

void func(int signo, siginfo_t *info, void *context);

typedef struct
{
    int     si_signo;   /* same value as signo argument */
    int     si_code;    /* SI_{USER, QUEUE, TIMER, ASYNCIO, MESGQ}*/
    union sigval si_value;    /* integer or pointer value from sender */
} siginfo_t;

程序1：创建一个消息队列，其名字是作为命令行参数指定，消息队列创建成功后输出队列的属性。程序如下：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include<sys/stat.h>
#define FILE_MODE (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)

int main(int argc,char *argv[])
{
    int     c,flags;
    mqd_t   mqd;
    struct mq_attr  attr;

    flags = O_RDWR|O_CREAT;
    printf("create mqueue.\n");
    while((c = getopt(argc,argv,"e")) != -1)
    {
        switch(c)
        {
            case 'e':
                flags |= O_EXCL;
                break;
        }
    }
    if(optind != argc-1)
    {
        perror("usage: mqcreate [-e] <name>");
        exit(0);
    }
    if((mqd = mq_open(argv[optind],flags,FILE_MODE,NULL)) == -1)
    {
        perror("mq_open() error");
        exit(-1);
    }
    mq_getattr(mqd,&attr);
    printf("max #msgs = %ld,max #bytes/msg = %ld,#currently on queue = %ld\n",
           attr.mq_maxmsg,attr.mq_msgsize,attr.mq_curmsgs);
    mq_close(mqd);
    exit(0);
}

编译运行:

gcc -o create\_msg create\_msg.c -lrt

./create_msg /msgqueue_test -e
create mqueue.
max #msgs = 10,max #bytes/msg = 8192,#currently on queue = 0

程序2：练习mq\_send和mq\_receive函数，调用mqsend程序向消息队列中写入消息，调用mqreceive程序从消息队列中读取消息。程序如下所示：

mq\_send.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include<sys/stat.h>
typedef unsigned int  uint_t;

int main(int argc,char *argv[])
{
    mqd_t   mqd;
    void    *ptr;
    size_t  len;
    uint_t  prio;
    if(argc != 4)
    {
        printf("usage: mqsend <name> <$bytes> <priority>\n");
        exit(0);
    }
    len = atoi(argv[2]);
    prio = atoi(argv[3]);
    mqd = mq_open(argv[1],O_WRONLY);
    ptr = calloc(len,sizeof(char));
    if(mq_send(mqd,ptr,len,prio) == -1)
    {
        perror("mq_send() error:");
        exit(-1);
    }
    exit(0);
}

mq\_receive.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <mqueue.h>
#include <fcntl.h>
#include <errno.h>
#include<sys/stat.h>
typedef unsigned int  uint_t;

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

    flags = O_RDONLY;
    while((c = getopt(argc,argv,"n")) != -1)
    {
        switch(c)
        {
            case 'n':
                flags |= O_NONBLOCK;  //设置为非阻塞
                break;
        }
    }
    if(optind != argc-1)
    {
        printf("usage: mqreceive [-n] <name>");
        exit(0);
    }
    mqd = mq_open(argv[optind],flags);
    mq_getattr(mqd,&attr);
    buff = malloc(attr.mq_msgsize);
    if((n = mq_receive(mqd,buff,attr.mq_msgsize,&prio)) == -1)
    {
        perror("mq_receive error: ");
        exit(-1);
    }
    printf("read %ld bytes,priority = %u\n",(long) n,prio);
    exit(0);
}

程序执行结果：

其中最后两条, 设置阻塞或者非阻塞从消息队列读取数据的情况.