Linux应用程序设计之消息队列

5.1、消息队列概述

消息队列就是一个消息的列表，用户可以从消息队列添加消息、读取消息等。从这点看，消息队列具有一定的FIFO的特性，但是它可以实现消息的随机查询，比FIFO具有更大的优势，同时，这些消息又是存在与内核中的，由"队列ID"来标识。

5.2、消息队列的实现

1、函数说明

消息队列的实现包括创建或打开消息队列、添加消息、读消息和控制消息队这四种操作。其中创建或打开消息队列使用的函数是msgget，这里创建的消息队列的数量会受到系统消息队列数量的限制；添加消息队列的函数是msgsnd函数，它把消息添加到以打开的消息队列末尾；读取消息使用的函数msgrcv，它把消息从消息队列中取走，与FIFO不同的是，这里可以指定取走某一条消息；最后控制消息队列使用的函数是msgctl，它可以完成多项功能。

2、函数格式

msgget函数：

所需头文件：#includ #include #include　

函数原型：int msgget(key_t key, int flag)

参数：key：返回新的或已有队列的队列ID，IPC_PRIVATE

flag：

返回值：成功：消息队列ID 出错： -1

msgsnd函数：

所需头文件：#includ #include #include　

函数原型：int msgsnd(int msqid,const void *ptr,size_t size, int flag)

参数：ptr：指向消息结构的指针。该消息结构msgbuf为

struct msgbuf{

long mtype;//消息类型

char mtext[1];//消息正文

}

size：消息的字节数，不要以NULL结尾

flag：IPC_NOWAIT若消息并没有立即发送而调用进程立即会返回

0：msgsnd调用阻塞知道条件满足为至。

返回值：成功：0 出错： -1

msgrcv函数：

所需头文件：#includ #include #include　

函数原型：int msgrcv(int msgid,struct msgbuf *msgp, int size,long msgtype,int flag)

参数：msqid：消息队列的队列ID

msgp：消息缓冲区

size：消息字节数，不要以null结尾

msgtype：0：接收消息队列中的第一个消息

大于0：接收消息队列中第一个类型为msgtype的消息

小于0：接收消息队列中的第一个类型值不小于msgtype绝对值且类型值又最小的消息

flag：MSG_NOERROR：若返回的消息比size字节多，则消息就会截断到size字节，且不通知消息发送进程。

IPC_NOWAIT：若消息并没有立即发送而调用进程会立即返回

0：msgsnd调用阻塞知道条件满足为止

返回值：成功：0 失败 -1

msgctl函数：

所需头文件：#includ #include #include　

函数原型：int msgctl(int msgqid , int cmd, struct msqid_ds *buf)

参数： msqid：消息队列的队列ID

cmd：IPC_STAT：读消息第一列的数据结构msqid_ds，并将其存储在buf指定的地址中。

IPC_SET：设置消息队列的数据结构msqid_ds中的ipc_perm元素的值，这个值取自buf参数

IPC_RMID：从系统内核中移走消息队列

Buf：消息队列缓冲区

返回值：成功：0 出错 -1

3、使用实例

这个实例体现了如何使用消息队列进行进程间通信，包括消息队列的创建，消息发送和读取，消息队列的撤销等多种操作，注意这里使用了函数fotk，它可以根据不同的路径和关表示产生标准的key。程序源代码如下所示：

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

#define BUFSZ 512

struct message{

long msg_type;

char msg_text[BUFSZ];
};

int main()
{
int qid;
key_t key;
int len;
struct message msg;

/*根据不同的路径和关键表示标准的key*/

if((key=ftok(".",'a')) == -1){
perror("ftok");
exit(1);
}

/*创建消息队列*/

if((qid = msgget(key, IPC_CREAT|0666)) == -1){
perror("msgget");
exit(1);
}

printf("opened queue %d\n",qid);
puts("Please enter the message to queue: ");
if((fgets((&msg)->msg_text,BUFSZ,stdin)) == NULL){
puts("no message");
exit(1);
}

msg.msg_type = getpid();
len = strlen(msg.msg_text);

/*添加消息队列*/
if((msgsnd(qid,&msg,len,0)) < 0){
perror("message posted");
exit(1);
}

/*读取消息队列*/
if(msgrcv(qid,&msg,BUFSZ,0,0)<0){
perror("msgrcv");
exit(1);
}
printf("message is : %s\n",(&msg)->msg_text);

/*从系统内核中移走消息队列*/
if((msgctl(qid,IPC_RMID,NULL)) < 0){
perror("msgctl");
exit(1);
}
exit(0);

}