【Linux进程间通信】 - 消息队列

原创作品，转载请标明：http://blog.csdn.net/Xiejingfa/article/details/50929937

今天我们来介绍一下如何使用消息队列来进行进程间通信。

1、消息队列

消息队列是由内核维护的一种链式结构。链表中每一个记录又称作消息，消息具有特定的格式和优先级别。各个进程通过消息队列标识符来引用消息队列，这样，写进程就可以按照一定的规则添加新的消息，读进程可以按一定的规则取走消息（具体按什么规则我们稍后讨论）。和前面介绍的共享内存和信号量一样，消息队列是随内核持续的，也就是说我们使用完毕后需要显式删除消息队列。

每一个消息队列都一个msqid_ds结构与之关联。用户可以使用该结构来设置或获取消息队列的相关信息。

在我使用的Ubuntu14.04系统中msqid_ds结构定义如下：

struct msqid_ds {
struct ipc_perm msg_perm;
struct msg *msg_first;      /* first message on queue,unused  */
struct msg *msg_last;       /* last message in queue,unused */
__kernel_time_t msg_stime;  /* last msgsnd time */
__kernel_time_t msg_rtime;  /* last msgrcv time */
__kernel_time_t msg_ctime;  /* last change time */
unsigned long  msg_lcbytes; /* Reuse junk fields for 32 bit */
unsigned long  msg_lqbytes; /* ditto */
unsigned short msg_cbytes;  /* current number of bytes on queue */
unsigned short msg_qnum;    /* number of messages in queue */
unsigned short msg_qbytes;  /* max number of bytes on queue */
__kernel_ipc_pid_t msg_lspid;   /* pid of last msgsnd */
__kernel_ipc_pid_t msg_lrpid;   /* last receive pid */
};

2、如何使用消息队列

在Linux中，与消息队列相关的系统调用主要有msgget、msgsnd、msgrcv、msgctl四个函数，这些函数的使用和前面介绍的共享内存和信号量的操作函数类似，我们可以进行类比学习。下面我们就具体介绍一下这些函数。

2.1、创建或打开消息队列

使用msgget函数来创建或打开一个消息队列。该函数的原型如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgget(key_t key, int msgflg);

如果调用成功，该函数返回消息队列的标识符ID，否则返回-1。

在msgget函数中，参数key我们已经在前面共享内存通信中详细介绍过，有关该参数的内容，可以参考这篇文章【Linux进程间通信】 - 共享内存

参数msgflg是一些标志位，常用的有IPC_CREAT、IPC_EXCL、IPC_NOWAIT三个取值。具体作用我们也在前面的文章中介绍过，这里就不在赘述。

到这里我们可以看到，共享内存、信号量、消息队列者三种进程间通信方式中的xxxget方法中的参数都包含了一个键值key参数和一个或一组或操作连接起来的标识flg参数。它们的含义基本相同。

如果msgget创建了一个新的消息队列，它相应的msqid_ds结构会被初始化。

2.2、消息队列的写操作

消息队列的写操作就是往消息队列中发送数据（消息），主要通过msgsnd函数来执行写操作。

在介绍msgsnd函数前，我们先来看看消息队列中消息的定义。

消息队列中所传递的消息一般由“消息的类型”和“消息数据”两部分组成，一般用msgbuf结构表示。该结构通常定义为：

struct msgbuf
{
long msgtype;
char msgtext[1024];
};

其中，msgtype通常是一个长整型，表示消息类型。设置消息类型字段具有重要作用，当我们从消息队列中读取消息时可以根据消息类型获取消息。而msgtext表示数据域，通常可以根据需要设置不同的大小。上面我们就定义了一个数据域长度为1024的消息。

msgsnd用来发送消息。发送消息的过程大致是：先声明一个msgbuf消息缓冲区，然后往里面填充消息，即设置消息类型和所需要传递的数据。最后调用msgsnd函数把消息发送到消息队列。msgsnd函数的原型如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

如果发送成功，msgsnd函数 返回0，否则返回-1。各个参数的具体含义如下：

（1）、第一个参数msqid

参数msqid是消息队列的标识符ID，也就是msgget函数的返回值。

（2）、第一个参数msgp

参数msgp是指向需要发送的消息msgbuf结构的指针。

（3）、第一个参数msgsz

参数msgsz表明消息msgbuf中数据缓冲区msgtext的长度，而不是整个消息的长度。

（4）、第一个参数msgflg

参数msgflg用来指定消息队列空间满了无法容纳其它消息时的处理方法。当消息队列满时，如果msgflg设置为IPC_NOTWAIT，则msgsnd立即出错返回EAGAIN。否则发送消息的进程被阻塞直到消息队列中有空间或者消息队列被删除或者调用进程捕获一个信号并从信号处理程序返回时msgsnd函数才返回。

2.3、消息队列的读操作

消息队列的读操作是指从消息队列中读取消息。读操作的一般过程是：首先声明一个msgbuf类型的消息，然后调用msgrcv函数把消息读入该缓冲区。

msgrcv函数的原型如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,
int msgflg);

如果操作成功，msgrcv函数返回消息的字节长度，否则返回-1。各参数的具体含义如下：

（1）第一个参数msqid

参数msqid是消息队列的标识符ID，也就是msgget函数的返回值。

（2）第二个参数msgp

参数msgp指向一个msgbuf结构的缓冲区，msgrcv操作成功后将消息放入msgp指定的消息缓冲区中。

（3）第三个参数msgsz

参数msgsz表明消息msgbuf中数据缓冲区msgtext的长度，而不是整个消息的长度。

（4）第四个参数msgtyp

参数msgtyp用来指明要接收的消息类型。如下表：

msgtyp的取值	具体含义
msgtyp = 0	读取消息队列中的第一条消息（先进先出）
msgtyp > 0	读取消息队列中消息类型等于msgtyp的第一条消息
msgtyp < 0	读取消息队列中消息类型小于等于msgtyp绝对值的所有消息中类型值最小的一条消息

（5）第五个参数msgflg

参数msgflg用来指定函数的操作类型。当读消息时，如果msgflg的值被设置为IPC_NOWAIT时，如果没有指定类型的消息可用，则msgrcv直接返回，errno被设置为ENOMSG。否则调用进程会一直阻塞直到消息队列中有消息可用，或者该消息队列被删除（此时errno = EIDRM），又或者调用进程捕获到一个信号并从信号处理程序中返回（此时errno = EINTR）时msgrcv函数才返回。

msgrcv函数调用成功后，与消息队列关联的msgid_ds结构也会得到更新。

2.4、消息队列的控制

我们可以使用smgctl函数来获得或设置消息队列的属性。smgctl函数的原型如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

如果调用成功，msgctl函数返回0，否则返回-1。各参数的含义如下：

（1）第一个参数msqid

参数msqid是消息队列的标识符ID，也就是msgget函数的返回值。

（2）、第二个参数cmd

参数cmd指定了操作类型，常用有三种命令：

cmd的取值	具体操作
IPC_STAT	获取消息队列的属性并将结果存入第三个参数指定的msgid_ds结构中
IPC_SET	根据第三个参数指定的msgid_ds结构设置消息队列的属性
IPC_RMID	删除msqid标识的消息队列

（3）、第三个参数buf

参数buf主要配合cmd参数使用。

3、实例演示

下面通过一个例子演示一下消息队列的读写操作。

写进程msg_write.c：

#include <unistd.h>
#include <sys/msg.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <string.h>

#define MSG_SIZE 512

struct msgbuf
{
long msgtype;
char msgtext[MSG_SIZE];
};

int main()
{
int msgid;

// 创建或打开消息队列
msgid = msgget((key_t) 2345, IPC_CREAT | 666);
if (msgid == -1)
{
perror("msgget errno!");
exit(1);
}

// 发送消息
struct msgbuf data;
data.msgtype = 1;
bzero(&data.msgtext, MSG_SIZE);
sprintf(data.msgtext, "Hello, I am msg_send.c, my pid is %u\n", getpid());
if (msgsnd(msgid, (void *) &data, MSG_SIZE, 0) == -1)
{
perror("msgsnd error!");
exit(1);
}

printf("send msg: %s\n", data.msgtext);

sleep(5);
}

读进程msg_read.c：

#include <unistd.h>
#include <sys/msg.h>
#include <sys/types.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>

#define MSG_SIZE 512

struct msgbuf
{
long msgtype;
char msgtext[MSG_SIZE];
};

int main()
{
int msgid;

// 创建或打开消息队列
msgid = msgget((key_t) 2345, IPC_CREAT | 666);
if (msgid == -1)
{
perror("msgget error!");
exit(1);
}

// 读取消息
struct msgbuf data;
bzero(data.msgtext, MSG_SIZE);
long type = 0;
if (msgrcv(msgid, (void *) &data, MSG_SIZE, type, 0) == -1)
{
perror("msgrcv error!");
exit(1);
}
printf("msg_rend.c read from msg queue: %s\n", data.msgtext);

sleep(5);

// 删除消息队列
if (msgctl(msgid, IPC_RMID, 0) == -1)
{
perror("msgctl error!");
exit(1);
}
};

对于上面的读进程和写进程，两则的运行顺序不同，运行结果也不同。

如果写进程先运行，写进程输出：

send msg: Hello, I am msg_send.c, my pid is 20237

接着运行读进程，读进程输出：

msg_rend.c read from msg queue: Hello, I am msg_send.c, my pid is 20237

如果读进程先运行，消息队列中没有消息，msgrcv函数会一直阻塞直到有消息可读。

4、消息队列通信的特点

与管道相比，无论匿名管道还是命名管道，它们只能承载无格式的字节流。而消息队列中的消息有固定的消息格式和消息类型，并且读进程可以通过制定消息类型来有选择的读取消息队列中的消息。

与共享内存相比，共享内存通信虽然效率高，但是需要使用额外的手段（如信号量）进行同步控制（这点命名管道也需要）。而消息队列却不需要担心同步问题。

参考：《Unix环境高级编程》