IPC 消息队列 一

消息队列可以认为是一个消息链表,某个进程往一个消息队列中写入消息之前,不需要另外某个进程在该队列上等待消息的达到,这一点与管道和FIFO相反。Posix消息队列与System V消息队列的区别如下:
1. 对Posix消息队列的读总是返回最高优先级的最早消息,对System V消息队列的读则可以返回任意指定优先级的消息。
2. 当往一个空队列放置一个消息时,Posix消息队列允许产生一个信号或启动一个线程,System V消息队列则不提供类似的机制。

Posix消息队列操作函数如下:

#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:创建一个消息队列,其名字是作为命令行参数指定,消息队列创建成功后输出队列的属性。程序如下:

本测试是用的Linux Ubuntu系统,编译程序的时候需要添加-lrt连接,如果不加会提示如下错误信息

因此正确编译的方式如下:

程序编译完成后,如果直接运行程序则提示mq_open失败,提示mq_open permission denied。解决办法是:

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

然后再运行即可看到创建的消息队列。程序结果如下所示:

程序2:练习mq_send和mq_receive函数,调用mqsend程序向消息队列中写入消息,调用mqreceive程序从消息队列中读取消息。程序如下所示:

mqsend程序:

 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <unistd.h>
 4 #include <mqueue.h>
 5 #include <fcntl.h>
 6 #include <errno.h>
 7 #include<sys/stat.h>
 8 typedef unsigned int  uint_t;
 9
10 int main(int argc,char *argv[])
11 {
12     mqd_t   mqd;
13     void    *ptr;
14     size_t  len;
15     uint_t  prio;
16     if(argc != 4)
17     {
18         printf("usage: mqsend <name> <$bytes> <priority>\n");
19         exit(0);
20     }
21     len = atoi(argv[2]);
22     prio = atoi(argv[3]);
23     mqd = mq_open(argv[1],O_WRONLY);
24     ptr = calloc(len,sizeof(char));
25     if(mq_send(mqd,ptr,len,prio) == -1)
26     {
27         perror("mq_send() error:");
28         exit(-1);
29     }
30     exit(0);
31 }

mqreceive程序:

 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <unistd.h>
 4 #include <mqueue.h>
 5 #include <fcntl.h>
 6 #include <errno.h>
 7 #include<sys/stat.h>
 8 typedef unsigned int  uint_t;
 9
10 int main(int argc,char *argv[])
11 {
12     int     c,flags;
13     mqd_t   mqd;
14     ssize_t n;
15     uint_t  prio;
16     void *buff;
17     struct mq_attr attr;
18
19     flags = O_RDONLY;
20     while((c = getopt(argc,argv,"n")) != -1)
21     {
22         switch(c)
23         {
24             case ‘n‘:
25                 flags |= O_NONBLOCK;  //设置为非阻塞
26                 break;
27         }
28     }
29     if(optind != argc-1)
30     {
31         printf("usage: mqreceive [-n] <name>");
32         exit(0);
33     }
34     mqd = mq_open(argv[optind],flags);
35     mq_getattr(mqd,&attr);
36     buff = malloc(attr.mq_msgsize);
37     if((n = mq_receive(mqd,buff,attr.mq_msgsize,&prio)) == -1)
38     {
39         perror("mq_receive error: ");
40         exit(-1);
41     }
42     printf("read %ld bytes,priority = %u\n",(long) n,prio);
43     exit(0);
44 }

程序执行结果如下所示:

程序3:信号通知函数使用,当有一个消息放置到某个空队列中,该程序产生信号,通知进程消息队列中放入了一个新的消息。程序如下:

 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <unistd.h>
 4 #include <mqueue.h>
 5 #include <fcntl.h>
 6 #include <errno.h>
 7 #include <signal.h>
 8
 9 typedef unsigned int  uint_t;
10
11 volatile    sig_atomic_t mqflag;  //全局变量,检查信号的产生
12 static void sig_usr1(int);
13
14 int main(int argc,char *argv[])
15 {
16     mqd_t       mqd;
17     void        *buff;
18     ssize_t     n;
19     sigset_t    zeromask,newmask,oldmask;
20     struct mq_attr  attr;
21     struct sigevent sigev;
22     if(argc != 2)
23     {
24         printf("usage :mqnotify <name>");
25         exit(0);
26     }
27     mqd = mq_open(argv[1],O_RDONLY);
28     mq_getattr(mqd,&attr);
29     buff = malloc(attr.mq_msgsize);
30     sigemptyset(&zeromask);
31     sigemptyset(&newmask);
32     sigemptyset(&oldmask);
33     sigaddset(&newmask,SIGUSR1);
34     signal(SIGUSR1,sig_usr1);
35     sigev.sigev_notify = SIGEV_SIGNAL;
36     sigev.sigev_signo = SIGUSR1;
37     if(mq_notify(mqd,&sigev) == -1)
38     {
39         perror("mq_notify error");
40         exit(-1);
41     }
42     for(; ;)
43     {
44         sigprocmask(SIG_BLOCK,&newmask,&oldmask);
45         while(mqflag == 0)
46             sigsuspend(&zeromask); //挂起,等待
47         mqflag = 0;
48         mq_notify(mqd,&sigev);
49         n = mq_receive(mqd,buff,attr.mq_msgsize,NULL);
50         printf("read %ld bytes\n",(long) n);
51         sigprocmask(SIG_UNBLOCK,&newmask,NULL);
52     }
53     eixt(0);
54 }
55
56 static void sig_usr1(int signo)
57 {
58     mqflag = 1;
59     return ;
60 }

程序执行结果如下:

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

 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <unistd.h>
 4 #include <mqueue.h>
 5 #include <fcntl.h>
 6 #include <errno.h>
 7 #include <signal.h>
 8
 9 int main(int argc,char *argv[])
10 {
11     mqd_t       mqd;
12     int         signo;
13     void        *buff;
14     ssize_t     n;
15     sigset_t    newmask;
16     struct mq_attr  attr;
17     struct sigevent sigev;
18     if(argc != 2)
19     {
20         printf("usage :mqnotify <name>");
21         exit(0);
22     }
23     mqd = mq_open(argv[1],O_RDONLY);
24     mq_getattr(mqd,&attr);
25     buff = malloc(attr.mq_msgsize);
26     sigemptyset(&newmask);
27     sigaddset(&newmask,SIGUSR1);
28     sigprocmask(SIG_BLOCK,&newmask,NULL);
29
30     sigev.sigev_notify = SIGEV_SIGNAL;
31     sigev.sigev_signo = SIGUSR1;
32     if(mq_notify(mqd,&sigev) == -1)
33     {
34         perror("mq_notify error");
35         exit(-1);
36     }
37     for(; ;)
38     {
39        sigwait(&newmask,&signo); //阻塞并等待该信号
40        if(signo == SIGUSR1)
41        {
42             mq_notify(mqd,&sigev);
43             while((n = mq_receive(mqd,buff,attr.mq_msgsize,NULL))>=0)
44                 printf("read %ld bytes\n",(long) n);
45             if(errno != EAGAIN)
46             {
47                 perror("mq_receive error");
48                 exit(-1);
49             }
50        }
51     }
52     eixt(0);
53 }

启动线程处理消息通知,程序如下:

 1 #include <stdio.h>
 2 #include <stdlib.h>
 3 #include <unistd.h>
 4 #include <mqueue.h>
 5 #include <fcntl.h>
 6 #include <errno.h>
 7 #include <signal.h>
 8
 9 mqd_t       mqd;
10 struct mq_attr  attr;
11 struct sigevent sigev;
12 static void notify_thread(union sigval);
13
14 int main(int argc,char *argv[])
15 {
16
17     if(argc != 2)
18     {
19         printf("usage :mqnotify <name>");
20         exit(0);
21     }
22     mqd = mq_open(argv[1],O_RDONLY | O_NONBLOCK);
23     mq_getattr(mqd,&attr);
24
25     sigev.sigev_notify = SIGEV_THREAD;
26     sigev.sigev_value.sival_ptr = NULL;
27     sigev.sigev_notify_function = notify_thread;
28     sigev.sigev_notify_attributes = NULL;
29
30     if(mq_notify(mqd,&sigev) == -1)
31     {
32         perror("mq_notify error");
33         exit(-1);
34     }
35     for(; ;)
36     {
37         pause();
38     }
39     eixt(0);
40 }
41 static void notify_thread(union sigval arg)
42 {
43     ssize_t     n;
44     void        *buff;
45     printf("notify_thread started\n");
46     buff = malloc(attr.mq_msgsize);
47     mq_notify(mqd,&sigev);
48     while((n = mq_receive(mqd,buff,attr.mq_msgsize,NULL))>=0)
49                 printf("read %ld bytes\n",(long) n);
50     if(errno != EAGAIN)
51     {
52                 perror("mq_receive error");
53                 exit(-1);
54     }
55     free(buff);
56     pthread_exit(NULL);
57 }
时间: 2024-10-12 06:57:42

IPC 消息队列 一的相关文章

IPC: 消息队列

#################################################### 消息队列   消息队列分为: 1.posix消息队列:可以在同一主机上有亲缘关系或无亲缘关系的进程间使用. 2.system v消息队列:同上. 消息队列有随内核的持续性. ----------------------------------------------------------- posix消息队列: gcc -lrt #include <mqueue.h> #include

IPC——消息队列双向通行

消息队列提供了一个进程向另一个进程发送数据块的方法,每个数据块都被认为是有一个类型的,这个类型下文中是用常量is_client_snd和is_server_snd来表示的 消息队列相比管道来说的优点是避免了阻塞. 系统调用函数: #include<sys/types.h> #include<sys/ipc.h> 原型:key_t ftok(const char* pathname,int proj_id); 参数:pathname为一个已存在的.可获得信息的文件的全路径(必须是已经

IPC——消息队列

Linux进程间通信——使用消息队列 下面来说说如何用不用消息队列来进行进程间的通信,消息队列与命名管道有很多相似之处.有关命名管道的更多内容可以参阅我的另一篇文章:Linux进程间通信——使用命名管道 一.什么是消息队列 消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法.消息队列是消息的链接表,存放在内核中并由消息队列标识符标识.  每个数据块都被认为含有一个类型,接收进程可以独立地接收含有不同类型的数据结构.我们可以通过发送消息来避免命名管道的同步和阻塞问题(命名管道要读端和写端

Linux进程通信(二)IPC消息队列

一.什么是消息队列 消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法.每个数据块都被认为含有一个类型,接收进程可以独立地接收含有不同类型的数据结构.我们可以通过发送消息来避免命名管道的同步和阻塞问题.但是消息队列与命名管道一样,每个数据块都有一个最大长度的限制. Linux用宏MSGMAX和MSGMNB来限制一条消息的最大长度和一个队列的最大长度MSGMNI来限制消息队列的总数. 二.IPC对象数据结构 内核为每个IPC对象维护一个数据结构(/usr/include/linux/ip

Linux IPC 消息队列

1. Posix 消息队列 /* mq_open - open a message queue */#include <fcntl.h> /* For O_* constants */ #include <sys/stat.h> /* For mode constants */ #include <mqueue.h> mqd_t mq_open(const char *name, int oflag); mqd_t mq_open(const char *name, i

进程间通信IPC:消息队列,信号量,共享内存

2015.3.4星期三 阴天 进程间通信:IPC 文件对象:记录文件描述符,文件开关等 IPC标示符:系统全局的流水号两个进程要通信,打开的是唯一的对象进行通讯,通过key操作 XSI IPC:消息队列,信号量,共享内存. ipcs 查看ip对象共享内存,信号量,消息队列等信息ipcrm 删除一个IP对象 Linux为用户提供了完善的,强大的网络功能完善的内置网络:其他操作系统不包含如此紧密的和内核结合在一起的网络部分 共享内存标示符的获取有两种方法:ftok(pathname,id)另一个是K

System V IPC 之消息队列

消息队列和共享内存.信号量一样,同属 System V IPC 通信机制.消息队列是一系列连续排列的消息,保存在内核中,通过消息队列的引用标识符来访问.使用消息队列的好处是对每个消息指定了特定消息类型,接收消息的进程可以请求接收下一条消息,也可以请求接收下一条特定类型的消息. 相关数据结构 与其他两个 System V IPC 通信机制一样,消息队列也有一个与之对应的结构,该结构的定义如下: struct msqid_ds { struct ipc_perm msq_perm; struct m

Chromium的IPC消息发送、接收和分发机制分析

由于Chromium采用多进程架构,因此会涉及到进程间通信问题.通过前面一文的学习,我们知道Browser进程在启动Render进程的过程中会建立一个以UNIX Socket为基础的IPC通道.有了IPC通道之后,接下来Browser进程与Render进程就以消息的形式进行通信.我们将这种消息称为IPC消息,以区别于线程消息循环中的消息.本文就分析Chromium的IPC消息发送.接收和分发机制. 老罗的新浪微博:http://weibo.com/shengyangluo,欢迎关注! Chrom

第15章 进程间通行 15.6 XSI IPC 15.7 消息队列

15.6 XSI IPC (1)3种称作XSI IPC的IPC是: 1)消息队列 2)信号量 3)共享存储器 (2)标识符和键 1)标识符:是一个非负整数,用于引用IPC结构.是IPC对象的内部名. 2)键:IPC对象的外部名.可使多个合作进程能够在同一IPC对象上汇聚. (3)IPC_PRIVATE键: 用于创建一个新的IPC结构.不能指定此键来引用一个现有的IPC结构. (4)ftok函数: 由一个路径名和项目ID产生一个键. (5)ipc_perm结构体 规定了ipc结构的权限和所有者.