Linux进程之间通信消息队列

消息队列就是一些消息的列表，用户可以在消息队列中添加消息

和读取消息。消息队列具有一定的FIFO特性，但是它可以实现

消息的随机查询，比FIFO具有更大的优势。同时这些消息又

存在于内核中，由“队列”ID来标识

消息队列的实现包括创建或打开消息队列，添加消息，读取

消息和控制消息队列这四种操作。

1. 创建或打开消息队列，msgget()，创建消息队列的数量会受

到系统消息队列数量的限制

msgget(创建或打开消息队列)

表头文件 #include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

函数定义 int msgget(key_t key, int msgflg);

函数说明 key:和消息队列关联的key值,IPC_PRIVATE或ftok的返回值

msgflag:消息队列的访问权限

成功：返回消息队列ID，错误返回-1

linux_c# ipcs -q //查看消息队列

------ Message Queues --------

key msqid owner perms used-bytes messages

2. 添加消息队列，msgsnd()它把消息添加到已打开的消息队列未尾

头文件 #include <sys/types.h>

#include <sys/ipc.h>

#include <sys/shm.h>

函数定义 int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg)

msgqit:消息队列的队列ID

msgp:指向消息结构的指针，该消息结构msgbuf通常为：

struct msgbuf

{

long mtype;//消息类型，该结构必须从这个域开始

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

};

msgsz:消息正文的字节数（不包含消息类型指针变量）

msgflg:IPC_NOWAIT若消息无法立即发送（比如：当前消息队列已满）

函数会立即返回

0：msgsnd调用阻塞直到发送成功为止

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

3.msgrcv()读取消息队列，它把消息从消息队列中取走，与FIFO不同的是，这里可以

取走指定的一条消息

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);

函数说明 参数msgid:消息队列的ID，msgp:接收消息的缓冲区，msgsz:要接收消息的字节数注意类型一致

msgtype：

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

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

小于0：接收消息队列中类型值不小于msgtyp的绝对值且类型值又最小的消息。

msgflag：

0：若无消息函数会一直阻塞

IPC_NOWAIT：若没有消息，进程会立即返回ENOMSG。

4. msgctl()控制消息队列

msgctl(删除、获取、设置消息队列)

表头文件: #include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

函数定义: int msgctl(int msqid, int cmd, struct msqid_ds *buf);

函数说明:msgid:消息队列ID

cmd: IPC_STAT：读取消息队列的属性，并将其保存在buf指向的缓冲区中。

IPC_SET：设置消息队列的属性。这个值取自buf参数。

IPC_RMID：从系统中删除消息队列。

buf:消息队列缓冲区

返回值：成功0，错误返回-1

The msqid_ds data structure is defined in <sys/msg.h> as follows:

struct msqid_ds {

struct ipc_perm msg_perm; /* Ownership and permissions

time_t msg_stime; /* Time of last msgsnd() */

time_t msg_rtime; /* Time of last msgrcv() */

time_t msg_ctime; /* Time of last change */

unsigned long __msg_cbytes; /* Current number of bytes in

queue (non-standard) */

msgqnum_t msg_qnum; /* Current number of messages

in queue */

msglen_t msg_qbytes; /* Maximum number of bytes

allowed in queue */

pid_t msg_lspid; /* PID of last msgsnd() */

pid_t msg_lrpid; /* PID of last msgrcv() */

};

-------------------------------------------------------

例：使用两个消息队列进行两个进程（发送和接收端）之间的通信，包括消息队列

的创建，消息发送与读取，消息队列的撤销和删除多种操作

消息发送端进程和消息接收端进程之间不需要额外实现进程之间的同步。

函数ftok(),它可以根据不同的路径和关键字产生标准的key

----------------------------------------------------------
msg_com.h
-----------------------------------------------------------
#ifndef __MSG_COM_H__
#define	__MSG_COM_H__

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

#define BUFFER_SIZE 512

struct message
{
long msg_type;//消息类型
char msg_text[BUFFER_SIZE];//消息正文
};
#endif
-----------------------------------------------------
msgsnd.c

#include "msg_com.h"
/**
*消息队列发送端
*/
int main(int argc, char **argv)
{
int qid;
key_t key;
struct message msg;
//根据不同的路径和关键字产生标准的key
if ((key = ftok(".", 'c')) == -1){
perror("ftok");
exit(1);
}

//创建消息队列
qid = msgget(key, IPC_CREAT | 0666);
if (qid == -1){
perror("msgget");
exit(1);
}
printf("Open queue %d\n", qid);

while(1) {
printf("Enter some message to the queue(quit to exit):\n");
if(fgets(msg.msg_text, BUFFER_SIZE, stdin) == NULL) {
puts("fgets");
exit(1);
}
msg.msg_type = getpid();
//添加消息到消息队列
if((msgsnd(qid, &msg, strlen(msg.msg_text), 0)) < 0) {
perror("msgsnd");
exit(1);
}

if (strncmp(msg.msg_text, "quit", 4) == 0) {
break;
}
}

exit(0);
}
-----------------------------------------------------
msgrcv.c

#include "msg_com.h"
/**
*消息队列接收端
*/
int main(int argc, char **argv)
{
int qid;
key_t key;
struct message msg;

if ((key=ftok(".", 'c')) == -1){
perror("ftok");
exit(1);
}
//创建消息队列
if ((qid = msgget(key, IPC_CREAT|0666)) == -1){
perror("msgget");
exit(1);
}
printf("Open queue %d\n", qid);
do {
//读取消息队列
memset(msg.msg_text, 0, BUFFER_SIZE);
if (msgrcv(qid, (void *)&msg, BUFFER_SIZE, 0, 0) < 0) {
perror("msgrcv");
exit(1);
}
printf("the message from process %ld: %s\n", msg.msg_type, msg.msg_text);

} while(strncmp(msg.msg_text, "quit", 4));
//从系统内核中移走消息队列
if ((msgctl(qid, IPC_RMID, NULL)) < 0) {
perror("msgctl");
exit(1);
}
exit(0);
}
--------------------------------------------------------------------
测试结果
msgque# ./msgsnd
Open queue 163840
Enter some message to the queue(quit to exit):
good day
Enter some message to the queue(quit to exit):
day is good day
Enter some message to the queue(quit to exit):
time is money
Enter some message to the queue(quit to exit):
quit
msgque# ./msgrcv
Open queue 163840
the message from process 6150: good day

the message from process 6150: day is good day

the message from process 6150: time is money

the message from process 6150: quit