Linux进程间通讯之消息队列

分手后的思念是犯贱 2023-10-17 10:49 12阅读 0赞

1. 基本概念

消息队列的最佳定义是：内核地址空间中的内部链表。消息可以顺序地发送到队列中，

并以几种不同的方式从队列中获取。当然，每个消息队列都是由 IPC标识符所唯一标识的。

2. 内部和用户数据结构

要完成理解象系统 V IPC这样复杂的问题，关键是要彻底熟悉内核的几个内部数据结构。

甚至对那些最基本的操作来说，直接访问这些结构中的某几个结构也是必要的，而其他的结

构则停留在一个更低的级别上。

3. 消息缓冲区

我们要介绍的第一个结构是 msgbuf结构。这个特殊的数据结构可以认为是消息数据的模

板。虽然定义这种类型的数据结构是程序员的职责，但是读者绝对必须知道实际上存在

msgbuf类型的结构。它是在在 linux/msg.h中定义的，有2个成员：

• mtype—它是消息类型，以正数来表示。这个数必须为一个正数！

• mtext—它就是消息数据。

4. 内核msg结构

内核把消息队列中的每个消息都存放在 msg结构的框架中。该结构是在 linux/msg.h中定义

的，如下是其成员的描述：

• msg\_next—这是一个指针，指向消息队列中的下一个消息。在内核寻址空间中，它们

是当作一个链表存储的。

• msg\_type—这是消息类型，它的值是在用户结构 msgbuf中赋予的。

• msg\_spot—这是一个指针，指向消息体的开始处。

• msg\_ts—这是消息文本(消息体)的长度。

• 内核 msgid\_ds结构—IPC对象分为三类，每一类都有一个内部数据结构，该数据结构

是由内核维护的。对于消息队列而言，它的内部数据结构是 msqid\_ds结构。对于系统上

创建的每个消息队列，内核均为其创建、存储和维护该结构的一个实例。该结构在

linux/msg.h中定义，如下所示：

struct msqid\_ds\{

struct ipc\_perm msg\_perm;

msgqnum\_t msg\_qnum;

msglen\_t msg\_qbytes;

pid\_t msg\_lspid;

pid\_t msg\_lrpid;

time\_t msg\_stime;

time\_t msg\_rtime;

time\_t msg\_ctime;

...

\};

在不同的系统中，此结构会有不同的新成员，这里只列出最少拥有的关键成员。其中，msg\_qbytes成员以及msg\_qnum成员在不同的系统也会有不同的上限值，这里就不逐一介绍了，详细内容请参阅相关系统手册。

• msg\_perm—它是 ipc\_perm结构的一个实例， ipc\_perm结构是在 linux/ipc.h中定义的。

该成员存放的是消息队列的许可权限信息，其中包括访问许可信息，以及队列的创建者

的有关信息 (如uid等等)。

• msg\_first—链接到队列中的第一个消息 (列表头部 )。

• msg\_last—链接到队列中的最后一个消息 (列表尾部)。

• msg\_stime—发送到队列的最后一个消息的时间戳 (time\_t)。

• msg\_rtime—从队列中获取的最后一个消息的时间戳。

• msg\_ctime—对队列进行最后一次变动的时间戳。

5. 内核ipc\_perm结构

内核把IPC对象的许可权限信息存放在 ipc\_perm类型的结构中。例如在前面描述的某个消

息队列的内部结构中， msg\_perm成员就是 ipc\_perm类型的，它的定义是在文件 linux/ipc.h中，

以上所有的成员都具有相当的自扩展性。对象的创建者以及所有者 (它们可能会有不同 )的

有关信息，以及对象的 IPC关键字都是存放在该结构中的。八进制形式的访问模式也是存放在

这里的，它是以一种无符号短整型的形式存储的。最后，时间片使用序列编号存放在最后面，

每次通过系统调用关闭 IPC对象(摧毁)时，这个值将被增加一，至多可以增加到能驻留在系统

中的IPC对象的最大数目。用户需要关心这个值吗？答案是“不”。

有关这个问题，在 Richard Stevens所著的《 Unix Network Programming 》一书的第125页

中作了精辟的讨论。该书还介绍了 ipc\_perm结构的存在和行为在安全性方面的原因。

6. 创建消息队列：

（1） msgget简介：为了创建一个新的消息队列，或者访问一个现有的队列，可以使用系统调用 msgget ( )。

msgget ( )的第一个变元是关键字的值 (在我们的例子中该值是调用 ftok ( )的返回值)。这个

关键字的值将被拿来与内核中其他消息队列的现有关键字值相比较。比较之后，打开或者访

问操作依赖于msgflg变元的内容。

• IPC\_CREAT—如果在内核中不存在该队列，则创建它。

• IPC\_EXCL—当与IPC\_CREAT一起使用时，如果队列早已存在则将出错。

如果只使用了 IPC\_CREAT, msgget ( )或者返回新创建消息队列的消息队列标识符，或者

会返回现有的具有同一个关键字值的队列的标识符。如果同时使用了

I P C \_ E X C L 和IPC\_CREAT，那么将可能会有两个结果。或者创建一个新的队列，或者如果该队列存在，则

调用将出错，并返回－1。IPC\_EXCL本身是没有什么用处的，但在与IPC\_CREAT组合使用时，

它可以用于保证没有一个现存的队列为了访问而被打开。

有个可选的八进制许可模式，它是与掩码进行 OR操作以后得到的。这是因为从功能上讲，

每个IPC对象的访问许可权限与 Unix文件系统的文件许可权限是相似的！

（2）msgget举例：

下面实例演示了使用msgget函数创建一个队列，函数中参数falgs指定为IPC\_CREAT|0666，说明新建一个权限为0666的消息队列，其中组用户、当前用户以及其他用户拥有读写的权限。并在程序的最后使用shell命令ipcs –q来查看系统IPC的状态。

（1）在vi编辑器中编辑该程序如下：

程序清单14-12 create\_msg.c msgget函数

\#include <sys/msg.h>

\#include <sys/types.h>

\#include <sys/ipc.h>

\#include <stdio.h>

\#include <stdlib.h>

int main ( void )

int qid;

key\_t key;

key = 113;

qid=msgget( key, IPC\_CREAT | 0666 )； /\*创建一个消息队列\*/

if ( qid < 0 ) \{ /\* 创建一个消息队列失败 \*/

perror ( "msgget" );

exit (1) ;

printf ("created queue id : %d /n", qid ); /\* 输出消息队列的ID \*/

system( "ipcs -q" ); /\*查看系统IPC的状态\*/

exit ( 0 );

（2）在shell中编译该程序如下：

$gcc create\_msg.c–o create\_msg

（3）在shell中运行该程序如下：

$./ create\_msg

created queue id : 0

\------ Message Queues --------

key msqid owner perms used-bytes messages

0x0000af40 623430 root 666 0 0

0x0000007b 0 root 666 0 0

在程序中使用了系统命令ipcs，命令参数\-q说明只查看消息队列的状态。注意在输出消息中，key段标明的是IPC的key值，msqid为该队列的ID值，perms为执行权限。同样，队列的执行权限像其他IPC对象一样没有执行权限。函数msgctl可以在队列上做多种操作，函数原型如下：

\#include <sys/msg.h>

int msgctl( int msqid, int cmd , struct msqid\_ds \*buf );

参数msqid为指定的要操作的队列，cmd参数指定所要进行的操作，其中有些操作需要buf参数。cmd参数的详细取值及操作如表14-9所示。

表14-9 cmd参数详解

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:60.9pt;"> cmd </td> 
 <td style="vertical-align:top;width:347.85pt;"> 操&nbsp;&nbsp;&nbsp; 作 </td> 
 </tr> 
 <tr> 
 <td style="width:60.9pt;"> IPC_STAT </td> 
 <td style="width:347.85pt;"> 取队列的msqid_ds结构，将它存放在buf所指向的结构中（需要buf参数） </td> 
 </tr> 
 <tr> 
 <td style="width:60.9pt;"> IPC_SET </td> 
 <td style="width:347.85pt;"> 使用buf所指向结构中的值对当前队列的相关结构成员赋值，其中包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_perm.cuid。该命令只能由具有以下条件的进程执行：进程有效用户ID等于msg_perm.cuid或msg_perm.uid超级用户进程。其中只有超级用户才可以增加队列的msg_qbytes的值 </td> 
 </tr> 
 <tr> 
 <td style="width:60.9pt;"> IPC_RMID </td> 
 <td style="width:347.85pt;"> 删除队列，并清除队列中的所有消息。此操作会影响后续进程对这个队列的相关操作。该命令只能由具有以下条件的进程执行。进程有效用户ID等于msg_perm.cuid或msg_perm.uid，超级用户进程 </td> 
 </tr> 
 </tbody> 
</table>

下面实例演示了调用msgctl函数操作队列，程序中先读取命令行参数，如没有，则打印命令提示信息，在调用msgctl函数执行删除操作的前后分别调用了一次shell命令ipcs –q来查看系统IPC的状态。

（1）在vi编辑器中编辑该程序如下：

程序清单14-13 del\_msg.c 调用msgctl删除指定队列

\#include <sys/msg.h>

\#include <sys/types.h>

\#include <sys/ipc.h>

\#include <stdio.h>

\#include <stdlib.h>

int main ( int argc ,char \*argv\[\] )

int qid ;

if ( argc != 2 )\{ /\* 命令行参数出错 \*/

puts ( "USAGE: del\_msgq.c <queue ID >" );

exit ( 1 );

qid = atoi ( argv\[1\] ); /\* 通过命令行参数得到组ID \*/

system( "ipcs -q");

if ( ( msgctl( qid, IPC\_RMID, NULL ) ) < 0 )\{ /\* 删除指定的消息队列 \*/

perror ("msgctl");

exit (1 );

system( "ipcs -q");

printf ( "successfully removed %d queue/n", qid ); /\* 删除队列成功 \*/

exit( 0 );

（2）在shell中编译该程序如下：

$gcc del\_msg.c–o del\_msg

（3）在shell中运行该程序如下：

$./ del\_msg

\------ Message Queues --------

key msqid owner perms used-bytes messages

0x0000007b 0 root 666 0 0

\------ Message Queues --------

key msqid owner perms used-bytes messages

successfully removed 0 queue

7. 读写消息队列

一旦获得了队列标识符，用户就可以开始在该消息队列上执行相关操作了。为了向队列

传递消息，用户可以使用 msgsnd系统调用.

由于消息队列的特殊性，系统为这个数据类型提供了两个接口（msgsnd函数，msgrcv函数），分别对应写消息队列及读消息队列。将一个新的消息写入队列，使用函数msgsnd，函数原型如下：

\#include <sys/msg.h>

int msgsnd ( int msqid, const void \*prt, size\_t nbytes, int flags);

对于写入队列的每一个消息，都含有三个值，正长整型的类型字段、数据长度字段和实际数据字节。新的消息总是放在队列的尾部，函数中参数msqid指定要操作的队列，ptr指针指向一个msgbuf的结构，定义如下：

struct msgbuf\{

long mtype;

char mbuf\[\];

\};

这是一个模板的消息结构，其中成员 mbuf是一个字符数组，长度是根据具体的消息来决定的，切忌消息不能以NULL结尾。成员mtype是消息的类型字段。

函数参数nbytes指定了消息的长度，参数flags指明函数的行为。函数成功返回0，失败返回–1并设置错误变量errno。errno可能出现的值有：EAGAIN、EACCES、EFAULT、EIDRM、EINTR、EINVAL和ENOMEM。当函数成功返回后会更新相应队列的msqid\_ds结构。

使用函数msgrcv可以从队列中读取消息，函数原型如下：

\#include <sys/msg.h>

ssize\_t msgrcv ( int msqid, void \*ptr, size\_t nbytes, long type , int flag);

函数中参数msqid为指定要读的队列，参数ptr为要接收数据的缓冲区，nbytes为要接收数据的长度，当队列中满足条件的消息长度大于nbytes的值时，则会参照行为参数flag的值决定如何操作：当flag中设置了MSG\_NOERROR位时，则将消息截短到nbytes指定的长度后返回。如没有MSG\_NOERROR位，则函数出错返回，并设置错误变量errno。设置type参数指定msgrcv函数所要读取的消息，tyre的取值及相应操作如表14-10所示。

表14-10 type值详解

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:112.7pt;"> type </td> 
 <td style="vertical-align:top;width:298.15pt;"> 操&nbsp;&nbsp;&nbsp; 作 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:112.7pt;"> 等于0 </td> 
 <td style="vertical-align:top;width:298.15pt;"> 返回队列最上面的消息（根据先进先出规则） </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:112.7pt;"> 大于0 </td> 
 <td style="vertical-align:top;width:298.15pt;"> 返回消息类型与type相等的第1条消息 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:112.7pt;"> 小于0 </td> 
 <td style="vertical-align:top;width:298.15pt;"> 返回消息类型小于等于type绝对值的最小值的第1条消息 </td> 
 </tr> 
 </tbody> 
</table>

参数flag定义函数的行为，如设置了IPC\_NOWAIT位，则当队列中无符合条件的消息时，函数出错返回，errno的值为ENOMSG。如没有设置IPC\_NOWAIT位，则进程阻塞直到出现满足条件的消息出现为止，然后函数读取消息返回。

下面实例演示了消息队列在进程间的通信。程序中创建了一个消息的模板结构体，并对声明变量做初始化。使用msgget函数创建了一个消息队列，使用msgsnd函数向该队列中发送了一条消息。

（1）在vi编辑器中编辑该程序如下：

程序清单14-14 snd\_msg.c 调用msgsnd函数向队列中发送消息

\#include <sys/msg.h>

\#include <sys/types.h>

\#include <sys/ipc.h>

\#include <stdio.h>

\#include <stdlib.h>

struct msg\{ /\*声明消息结构体\*/

long msg\_types; /\*消息类型成员\*/

char msg\_buf\[511\]; /\*消息\*/

\};

int main( void ) \{

int qid;

int pid;

int len;

struct msg pmsg; /\*一个消息的结构体变量\*/

pmsg.msg\_types = getpid(); /\*消息类型为当前进程的ID\*/

sprintf (pmsg.msg\_buf,"hello!this is :%d/n/0", getpid() ); /\*初始化消息\*/

len = strlen ( pmsg.msg\_buf ); /\*取得消息长度\*/

if ( (qid=msgget(IPC\_PRIVATE, IPC\_CREAT | 0666)) < 0 ) \{ /\*创建一个消 
息队列\*/

perror ( "msgget" );

exit (1) ;

if ( (msgsnd(qid, &pmsg, len, 0 )) < 0 )\{ /\*向消息队列中发送消息\*/

perror ( "msgsn" );

exit ( 1 );

printf ("successfully send a message to the queue: %d /n", qid);

exit ( 0 ) ;

（2）在shell中编译该程序如下：

$gcc snd\_msg.c –o snd\_msg

（3）在shell中运行该程序如下：

$./ snd\_msg

successfully send a message to the queue 0

上述程序中，先定义了一个消息的结构体。该结构体中包含两个成员，long类型成员msg\_types是消息的类型，注意，在消息队列中是以消息类型做索引值来进行检索的。char类型数组存放消息。在程序中先声明了一个消息的结构体变量，并做相应初始化，然后使用了msgget函数创建一个消息队列，并将该消息发送到此消息队列中。以下是一个使用消息队列发送消息的程序。

下面实例演示了如何使用队列读取消息。在程序的开始部分，判断用户是否输入了目标消息队列ID，如果没有，则打印命令的帮助信息；如果用户输入了队列的ID，则从队列中取出该消息，并输出到标准输出。

（1）在vi编辑器中编辑该程序。

程序清单14-15 rcv\_msg.c 使用msgrcv函数从指定队列中读出消息

\#include <sys/msg.h>

\#include <sys/types.h>

\#include <sys/ipc.h>

\#include <stdio.h>

\#include <stdlib.h>

\#define BUFSZ 4096

struct msg\{ /\*声明消息结构体\*/

long msg\_types; /\*消息类型成员\*/

char msg\_buf\[511\]; /\*消息\*/

\};

int main( int argc, char \* argv\[\] ) \{

int qid;

int len;

struct msg pmsg;

if ( argc != 2 )\{ /\*\*/

perror ( "USAGE: read\_msg <queue ID>" );

exit ( 1 );

qid = atoi ( argv\[1\] ); /\*从命令行中获得消息队列的ID\*/

/\*从指定队列读取消息 \*/

len = msgrcv ( qid, &pmsg, BUFSZ, 0, 0 );

if ( len > 0 )\{

pmsg.msg\_buf\[len\] = '/0'; /\*为消息添加结束符\*/

printf ("reading queue id :%05ld/n", qid ); /\*输出队列ID\*/

/\*该消息类型就是发送消息的进程ID\*/

printf ("message type : %05ld/n", pmsg.msg\_types );

printf ("message length : %d bytes/n", len ); /\*消息长度\*/

printf ("mesage text: %s/n", pmsg.msg\_buf); /\*消息内容\*/

else if ( len == 0 )

printf ("have no message from queue %d/n", qid );

else \{

perror ( "msgrcv");

exit (1);

system("ipcs -q")

exit ( 0 ) ;

（2）在shell中编译该程序如下：

$gcc rcv\_msg.c–o rcv \_msg

（3）在shell中运行该程序如下：

$./ rcv\_msg 0

reading queue id :0

message type : 03662

message length : 20 bytes

mesage text: hello!this is :3662

\------ Message Queues --------

key msqid owner perms used-bytes messages

0x00000000 0 root 666 0 0

该程序中声明了一个消息的结构体类型变量，并从命令行中得到所要操作的消息队列，然后使用函数msgrcv从指定消息队列中读取队列中最上面的一条消息（函数的第4个参数等于0，说明根据先进先出规则，应从队列的最上面读取一条消息），并将该消息输出到标准输出。在发送消息的程序中，消息类型字段指定的是发送消息进程的ID，可以使用该内容来判断信息的来源。