Posix消息队列.doc

Linux3+1暑期学习总结三(Posix消息队列)王晶晶一，消息队列简介： 消息队列可以看作是一个消息连表。它具有随内核的持续性，即当使用该消息队列的进程结束，或者已关闭该消息队列，该队列中的消息不会随之消失，只有在内核重新初始化，即计算机重启之后才会消失，因此称为随内核的持续性，这点也是与管道和FIFO的区别。消息队列的另一个特性是，在某一个进程往消息队列写消息之前不需要另外某个进程在该消息队列上等待消息的到达，即不会像管道个FIFO那样，如果往管道或者FIFO中些数据时，如果没有一个进程已经将读端打开，那么写操作会被阻塞。当然，如果从消息队列读取数据时，消息队列为空是会阻塞的。 每个消息都是一条记录，它有发送者赋予一个优先权，值越大优先级越高。 下图为一个消息队列可能的布局：头消息mq_maxmsgmq_msgsize下一个消息优先级30长度1数据下一个消息优先级20长度2数据NULL优先级10长度3数据该链表的头中含有当前队列的两个属性：队列中允许的最大消息数，每个消息的最大大小。二，相关函数解释：1 mq_open 所在头文件:#include函数原型：mqd_t mq_open(const char *name, int 0flag,.mode_t mode, struct mq_attr *attr)； 函数功能：创建消息队列。参数说明：name为消息队列的名字，根据消息队列的规则，为了更好的可移植性，该名字必须以开头，创建一个消息队列的时候无须路径，给出名字就好，其存放位置可有自己指定(创建前后都可以，下面会讲到)。oflag：为O_RDONLY（只读），O_WRONLY(只写)，O_RDWR(可读可写)之一，可能安位或上O_CREATE,O_EXCL（当消息已存在时，返回EEXIST错误到errno中），O_NONBLOCK(设置非阻塞)。mode和attr参数是可选，但是当实际操作是创建一个新队列时，即O_CREATE已指定，且要求创建的消息队列不存在，mode和attr参数是需要的。mode：表示创建消息对列的权限。由S_IRUSR,S_IWUSR,S_IXUSR,S_IRGRP,S_IWGRP,S_IXGRP,S_IROTH,S_IWOTH,S_IXOTH相或组成或者写成0777(表示rwxrwxrwx)等用八进制表示也可以。attr:在linux内核源代码中struct_mqattr定义的源代码如下： 存放消息队列的属性。其中mq_flags为0，表示阻塞，为O_NONBLOCK为非阻塞。 函数返回值：在内核源代码中mqd_t类型的定义如下：typedef _kernel_mqd_t mqd_t;typedef int _kernel_mqd_t;若创建成功则返回消息队列的描述符，否则返回1。2mq_close函数：所在头文件:#include函数原型：int mq_close(mqd_t mqdes);函数功能：关闭已打开的消息队列，关闭后调用进程不可以再使用该描述符，但其消息队列并没有被删除。一个进程终止时，它的所有打开着的消息队列都关闭，就像调用了mq_close一样。参数说明：mqdes为消息队列的描述符，即消息队列创建成功后的返回值。返回值：成功返回0，失败返回-1。3.mq_unlink函数 所在头文件:#include 函数原型：int mq_unlink(const char *name); 函数作用：从系统中删除名为name的消息队列。但删除的只是我们可以在系统中看见的文件的名字，但文件本身并没有被从磁盘上删除，除非该名称是文件的最后一个链接，并且该文件已关闭，才会将该文件真正从磁盘上删除。即如果某前该详细队列的文件还在其他进程中打开，那么不会将其从磁盘上删除，又或者这是最后一个链接，但它还为关闭，即未执行ma_close操作，或打开它的进程为结束就执行mq_unlink，它也不会从磁盘上删除。 函数的参数：消息队列的名称，以开始。 函数的返回值：成功返回0，出错返回1。4mq_getattr函数 所在头文件：#include 函数原型：int mq_getattr(mqd_t mqdes, struct mq_attr *attr); 函数功能：获取mqdes指的消息队列的属性，存放到attr结构体中。 参数说明：mqdes为消息队列描述符，attr为上面解释的存放消息队列属性的结构体。 函数返回值：成功返回0，失败返回1。5mq_setattr函数： 所在头文件：#include 函数原型：int mq_setattr(mqd_t mqdes, const struct mq_attr *attr, struct mq_attr *oattr); 函数功能：设置消息队列的属性，但是只使用attr结构体中的mq_flags属性，以设置(O_NONBLOCK)或清除(0)非阻塞标志。该结构体的另外三个属性被忽略，每个队列的最大消息数和每个消息的最大字节数都只能在创建时设置，当前队列中的消息数是随传送消息和读取消息的操作改变的，只能读取不能设置。如果oattr非空，那么指定队列的先前属性(4个)全将返回到由该指针指向的结构体中。 参数说明：见上述函数功能中对attr和oattr的解释。 返回值：成功返回0，失败返回1。6mq_send函数： 所在头文件：#include 函数原型：int mq_send(mqd_t mqdes, const char *ptr, size_t len, unsigned int prio); 函数功能：给描述符mqdes指向的消息队列发送消息，大小为len，内容存放在ptr中，prio为优先级。 参数说明：mqdes为要发送消息的消息队列描述符，ptr为要发送的数据，len为消息的长度，prio为消息的优先级。 返回值：成功返回0，失败返回1。7mq_receive函数： 所在头文件：#include 函数原型：ssize_t mq_receive(mqd_t mqdes, char *ptr, size_t len, unsigned int *proip); 函数功能：从描述符mqdes指向的消息队列中读取消息存放ptr中。 参数说明：mqdes为要从中读取消息的消息队列的描述符，ptr为存放接受到的消息的指针，len为接受的最大长度，该值不能小于能加到该消息对列上的最大大小，如果len小于该值，就立即返回EMSGSIZE错误。 返回值：成功返回读取消息的内容的字节数，出错返回1。8mq_notify函数： 所在头文件：#include,#include 函数原型：int mq_notify(mqd_t mqdes, const struct sigevent *notification); 函数功能：为指定队列建立或删除异步事件通知。 参数说明：在中： typedef union sigval int sival_int; void _user *sival_ptr; sigval_t;typedef struct sigevent sigval_t sigev_value; int sigev_signo; int sigev_notify; void (*sigev_notify_function)(union sigval); Pthread_attr_t *sigev_notify_attributes; sigevent_t;(1).如果notification参数非空，那么当前进程希望在有一个消息到达所指定的先前为空的队列时得到通知。我们说“该进程被注册为接收该队列的通知”。(2).如果notification参数为空指针，而且当前进程目前被注册为接收所指定队列的通知，那么已存在的注册将被撤销。(3).任意时刻只有一个进程可以被注册为接收某个指定队列的通知。(4).当有一个消息到达某个先前为空的队列，而且已有一个进程被注册为接收该队列的通知时，只有在没有任何线程阻塞在该队列的mq_receive调用中的前提下，通知才会发出。这就是说，在mq_receive调用中的阻塞比任何通知的注册都优先。(5).当该通知被发送给它的注册进程时，其注册即被撤销。该进程必须再次调用mq_notify重新注册(如果想要的话)。 返回值：成功返回0，失败返回1。三，相关程序示例：mq_create.c#include#include#include#include#include#include#includestruct mq_attr attr;int main(int argc , char *argv)int c, flags;mqd_t mqd;flags = O_RDWR|O_CREAT;/emz表示命令行的选项，m和z带有参数，getopt在我博客有详解while(c = getopt(argc, argv, em:z:) != -1)switch(c)case e:flags |= O_EXCL; break;case m:attr.mq_maxmsg = atol(optarg); break;case z:attr.mq_msgsize = atol(optarg); break;if(optind != argc - 1)perror(optind error!n);if(attr.mq_maxmsg != 0 & attr.mq_msgsize = 0) | (attr.mq_maxmsg = 0 & attr.mq_msgsize != 0)perror(must specify both -m maxmsg and -z msgsize!n);mqd = mq_open(argvoptind, flags, 0644, (attr.mq_maxmsg != 0) ? &attr : NULL);if(mqd != -1)printf(create ok!n);mq_close(mqd); /mq_close只是关闭了描述符，并不删除消息队列elseprintf(create error!n);exit(0);运行过程及结果：从上可以看出，创建一个消息队列时，其最大消息数和每个消息的最大字节数不能超出，系统的msg_max和msgsize_max，否则创建失败。当然每个系统对应的这两个值都不同，可通过上面的cat路径进行查看。/mq1创建失败是因为在这之前已经创建了一个/mq1，e选项就是用来判断当创建的消息队列存在时，返回错误信息。编译时在最后一定要加上-lrt，表示链接。在消息队列的名字前一定要加/，这是Posix的规定，为了更好的可移植性。接下来我们要对消息队列进行挂载，只有挂载后你才能看见你的消息队列：可以看到在程序的最后调用的mq_close，但并未删除消息队列的名字，只是关闭了描述符，这是mq_close和mq_unlink的最大区别。mqunlink.c#include#include#include#include#includeint main(int argc, char *argv)int flag;if(argc != 2)perror(usage: maunlink);flag = mq_unlink(argv1);if(flag = -1)printf(fail!n);elseprintf(ok!n);exit(0);运行过程及结果：mqsend.c#include#include#include#include#include#include#include#includeint main(int argc, char *argv)mqd_t mqd;char *ptr;size_t len;unsigned int prio;if(argc != 4)perror(./mqsend mqueue_name size prio!n);len = atoi(argv2);prio = atoi(argv3);mqd = mq_open(argv1, O_WRONLY);ptr = (char *)calloc(len, sizeof(char);strcpy(ptr, wjj_xyd);mq_send(mqd, ptr, len, prio);exit(0);mqreceive.c#include#include#include#include#include#includeint main(int argc, char *argv)int c, flags;mqd_t mqd;ssize_t n;unsigned int prio;char *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)perror(mqreceive error!n);mqd = mq_open(argvoptind, flags);mq_getattr(mqd, &attr);buff = (char*)malloc(attr.mq_msgsize);n = mq_receive(mqd, buff, attr.mq_msgsize, &prio);printf(buff = %s, read %ld bytes, priority = %un,buff, (long)n, prio);exit(0);mqsen.c和mqreceive.c的运行过程和结果：下面往一个消息队列中写入多条消息，然后读取，会发现读取的结果是按优先级从高到低(因为mq_receive的priop指针不空)：下来是一个简单的信号通知的程序，通过mq_notify函数实现：mqnotifysigl.c:#include#include#include#include#include#includemqd_t mqd;void *buff;struct mq_attr attr;struct sigevent sigev;static void sig_usr1(int);int main(int argc, char *argv)if(argc != 2)perror(argument error!n);mqd = mq_open(argv1, O_RDONLY);mq_getattr(mqd, &attr); /获取属性buff = malloc(attr.mq_msgsize);signal(SIGUSR1, sig_usr1);/接收到SIGUSR1信号后，执行sig_usr1函数sigev.sigev_notify = SIGEV_SIGNAL;sigev.sigev_signo = SIGUSR1;mq_notify(mqd, &sigev); /注册for(; ;) printf(dddddn);sleep(3);/pause();exit(0);static void sig_usr1(int signo)ssize_t n;mq_notify(mqd, &sigev); /重新注册n = mq_receive(mqd, buff, attr.mq_msgsize, NULL);printf(SIGUSR1 received,read %ld bytes, buff %sn, (lo