Linux系统编程——进程间通信：消息队列

概述

消息队列提供了一种在两个不相关的进程之间传递数据的简单高效的方法，其特点如下：

1. 消息队列可以实现消息的随机查询。消息不一定要以先进先出的次序读取，编程时可以按消息的类型读取。
2. 消息队列允许一个或多个进程向它写入或者读取消息。
3. 与无名管道、命名管道一样，从消息队列中读出消息，消息队列中对应的数据都会被删除。
4. 每个消息队列都有消息队列标识符，消息队列的标识符在整个系统中是唯一的。
5. 消息队列是消息的链表，存放在内存中，由内核维护。只有内核重启或人工删除消息队列时，该消息队列才会被删除。若不人工删除消息队列，消息队列会一直存在于系统中。

消息队列常用操作函数如下：

代码语言：javascript复制

#include <sys/msg.h>
#include <sys/types.h>
#include <sys/ipc.h>
 
key_t ftok(const char *pathname, int proj_id);
int msgget(key_t key, int msgflg);
int msgrcv(int msqid, void *msg_ptr, size_t msg_sz, long int msgtype, int msgflg);
int msgsnd(int msqid, const void *msg_ptr, size_t msg_sz, int msgflg);
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

对于消息队列的操作，我们可以类比为这么一个过程：假如 A 有个东西要给 B，因为某些原因 A 不能当面直接给 B，这时候他们需要借助第三方托管（如银行），A 找到某个具体地址的建设银行，然后把东西放到某个保险柜里（如 1 号保险柜），对于 B 而言，要想成功取出 A 的东西，必须保证去同一地址的同一间银行取东西，而且只有 1 号保险柜的东西才是 A 给自己的。

对于上面的例子，涉及到几个比较重要的信息：地址、银行、保险柜号码。

只有同一个地址才能保证是同一个银行，只有是同一个银行双方才能借助它来托管，只有同一个保险柜号码才能保证是对方托管给自己的东西。

而在消息队列操作中，键（key）值相当于地址，消息队列标示符相当于具体的某个银行，消息类型相当于保险柜号码。

同一个键（key）值可以保证是同一个消息队列，同一个消息队列标示符才能保证不同的进程可以相互通信，同一个消息类型才能保证某个进程取出是对方的信息。

键（key）值

System V 提供的进程间通信机制需要一个 key 值，通过 key 值就可在系统内获得一个唯一的消息队列标识符。key 值可以是人为指定的，也可以通过 ftok() 函数获得。

需要的头文件：

代码语言：javascript复制

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

key_t ftok(const char *pathname, int proj_id);

功能：

代码语言：javascript复制

获取键（key）值

参数：

代码语言：javascript复制

pathname: 路径名

proj_id: 项目ID，非 0 整数(只有低 8 位有效)

返回值：

代码语言：javascript复制

成功：key 值

失败：-1

消息队列的创建

所需头文件：

代码语言：javascript复制

#include <sys/msg.h>

int msgget(key_t key, int msgflg);

功能：

代码语言：javascript复制

创建一个新的或打开一个已经存在的消息队列。不同的进程调用此函数，
只要用相同的 key 值就能得到同一个消息队列的标识符。

参数：

代码语言：javascript复制

key: ftok() 返回的 key 值

msgflg: 标识函数的行为及消息队列的权限，其取值如下：

IPC_CREAT：创建消息队列。

IPC_EXCL： 检测消息队列是否存在。

位或权限位：消息队列位或权限位后可以设置消息队列的访问权限，
格式和open() 函数的 mode_t 一样（open() 的使用请点此链接），但可执行权限未使用。

返回值：

代码语言：javascript复制

成功：消息队列的标识符

失败：-1

示例代码如下：

代码语言：javascript复制

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
 
int main(int argc, char *argv[])
{
	key_t key;
	int  msgqid;
	
	key = ftok(".", 2012); // key 值
	
	// 创建消息队列
	msgqid = msgget(key, IPC_CREAT|0666);
	
	return 0;
}

运行结果如下：

消息队列的读写操作

对于消息队列的读写，都是以消息类型为准。消息类型相当于保险柜号码，A 往 1 号保险柜放东西，对方想取出 A 的东西必须也是从 1 号保险柜里取。同理，某一进程往消息队列添加 a 类型的消息，别的进程要想取出这进程添加的信息也必须取 a 类型的消息。

在学习消息队列读写操作前，我们先学习消息队列的消息格式：

代码语言：javascript复制

typedef struct _msg
{
	long mtype;		 // 消息类型
	char mtext[100]; // 消息正文
	//…… ……          // 消息的正文可以有多个成员
}MSG;

消息类型必须是长整型的，而且必须是结构体类型的第一个成员，类型下面是消息正文，正文可以有多个成员（正文成员可以是任意数据类型的）。至于这个结构体类型叫什么名字，里面成员叫什么名字，自行定义，没有明文规定。

添加信息

所需头文件：

代码语言：javascript复制

#include <sys/msg.h>

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

功能：

代码语言：javascript复制

将新消息添加到消息队列。

参数：

代码语言：javascript复制

msqid： 消息队列的标识符。

msgp：  待发送消息结构体的地址。

msgsz： 消息正文的字节数。

msgflg：函数的控制属性，其取值如下：

0：msgsnd() 调用阻塞直到条件满足为止。

IPC_NOWAIT: 若消息没有立即发送则调用该函数的进程会立即返回。

返回值：

代码语言：javascript复制

成功：0

失败：-1

获取信息

所需头文件：

代码语言：javascript复制

#include <sys/msg.h>

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

功能：

代码语言：javascript复制

从标识符为 msqid 的消息队列中接收一个消息。一旦接收消息成功，则消息在消息队列中被删除。

参数：

代码语言：javascript复制

msqid：消息队列的标识符，代表要从哪个消息列中获取消息。

msgp： 存放消息结构体的地址。

msgsz：消息正文的字节数。

msgtyp：消息的类型。可以有以下几种类型：

msgtyp = 0：返回队列中的第一个消息。

msgtyp > 0：返回队列中消息类型为 msgtyp 的消息（常用）。

msgtyp < 0：返回队列中消息类型值小于或等于 msgtyp 绝对值的消息，如果这种消息有若干个，
则取类型值最小的消息。

注意：在获取某类型消息的时候，若队列中有多条此类型的消息，则获取最先添加的消息，即先进先出原则。

msgflg：函数的控制属性。其取值如下：

0：msgrcv() 调用阻塞直到接收消息成功为止。

MSG_NOERROR: 若返回的消息字节数比 nbytes 字节数多,则消息就会截短到 nbytes 字节，
且不通知消息发送进程。

IPC_NOWAIT: 调用进程会立即返回。若没有收到消息则立即返回 -1。

返回值：

代码语言：javascript复制

成功：读取消息的长度

失败：-1

消息队列的控制

所需头文件：

代码语言：javascript复制

#include <sys/msg.h>

int msgctl(int msqid, int cmd, struct msqid_ds *buf) 功能：

代码语言：javascript复制

对消息队列进行各种控制，如修改消息队列的属性，或删除消息消息队列。

参数：

代码语言：javascript复制

msqid：消息队列的标识符。

cmd：函数功能的控制。其取值如下：

IPC_RMID：删除由 msqid 指示的消息队列，将它从系统中删除并破坏相关数据结构。

IPC_STAT：将 msqid 相关的数据结构中各个元素的当前值存入到由 buf 指向的结构中。
相当于，把消息队列的属性备份到 buf 里。

IPC_SET：将 msqid 相关的数据结构中的元素设置为由 buf 指向的结构中的对应值。
相当于，消息队列原来的属性值清空，再由 buf 来替换。

buf：msqid_ds 数据类型的地址，用来存放或更改消息队列的属性。

返回值：

代码语言：javascript复制

成功：0

失败：-1

实践示例

写端示例代码如下：

代码语言：javascript复制

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
 
typedef struct _msg
{
	long mtype;
	char mtext[50];
}MSG;
 
int main(int argc, char *argv[])
{
	key_t key;
	int  msgqid;
	MSG msg;
	
	key = ftok("./", 2015); // key 值
	
	// 创建消息队列
	msgqid = msgget(key, IPC_CREAT|0666);
	if(msgqid == -1)
	{
		perror("msgget");
		exit(-1);
	}
	
	msg.mtype = 10;	// 消息类型
	strcpy(msg.mtext, "hello mike"); // 正文内容
	
	/* 添加消息
	msg_id:消息队列标识符
	&msg：消息结构体地址
	sizeof(msg)-sizeof(long)：消息正文大小
	0：习惯用0
	*/
	msgsnd(msgqid, &msg, sizeof(msg)-sizeof(long), 0);
	
	return 0;
}

读端示例代码如下：

代码语言：javascript复制

#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
 
typedef struct _msg
{
	long mtype;
	char mtext[50];
}MSG;
 
int main(int argc, char *argv[])
{
	key_t key;
	int  msgqid;
	
	
	key = ftok("./", 2015); // key 值
	
	// 创建消息队列
	msgqid = msgget(key, IPC_CREAT|0666);
	if(msgqid == -1)
	{
		perror("msgget");
		exit(-1);
	}
	
	MSG msg;
	memset(&msg, 0, sizeof(msg));
	
	/* 取出类型为 10 的消息
	msg_id:消息队列标识符
	&msg：消息结构体地址
	sizeof(msg)-sizeof(long)：消息正文大小
	(long)10：消息的类型
	0：习惯用0
	*/
	msgrcv(msgqid, &msg, sizeof(msg)-sizeof(long), (long)10, 0);
	printf("msg.mtext=%sn", msg.mtext); 
	
	// 把消息队列删除
	// IPC_RMID：删除标志位
	msgctl(msgqid, IPC_RMID, NULL);
	
	return 0;
}

运行结果如下：

消息队列CMQ版 kafka .net

0 人点赞