Linux编程基础 5.2：消息队列_linux 消息队列

3 消息队列

• 消息队列就是一些消息的列表，或者说是一些消息组成的队列；
• 消息队列与管道有些类似，消息队列可以认为是管道的改进版。相较于管道的先进先出准则，消息队列在读取时可以按照消息的类型进行读取，这也是消息队列的特点，它可以实现消息随机查询。消息发送时，需要将消息封装，然后添加到队列的末尾即可；而消息接收时，则可以根据需求进行选择的读取（读取即将封装的消息从队列中移除）;
• 消息队列的本质是一个存放消息的链表，该链表由内核来维护。一个消息队列由一个标识符（即队列key）来标识。
• 消息队列的通信机制传递的数据具有某种结构，而不是简单的字节流；
• 向消息队列中写数据，实际上是向这个数据结构中插入一个新结点；
• 从消息队列中读数据，实际上是从这个数据结构中删除一个结点；
• 消息队列提供了一个从一个进程向另外一个进程发送一块数据的方法；
• 消息队列具有和管道一样的不足，每个数据块的最大长度是有上限的，系统上全体队列的最大总长度也是有上限的。

进程可以通过消息队列添加消息，同时也可以从消息队列中读取消息，它不同于无名管道的单向通信，操作更加灵活，如图所示：

使用消息队列实现进程间通信的步骤如下：

• （1）创建消息队列；
• （2）发送消息到消息队列；
• （3）从消息队列中读取数据；
• （4）删除消息队列。

Linux内核提供了4个系统调用：

key_t ftok(const char *pathname, int proj_id);//为队列随机附加key，pathename为路径，id号可随意（1-255）
int msgget(key_t key, int msgflg); //创建消息队列，返回值为该队列号
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);//发送消息
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);//接受信息，msgtyp需要指明接受的数据type
int msgctl(int msqid, int cmd, struct msqid_ds *buf);//对指定消息队列进行控制
1
2
3
4
5

3.1 ftok函数

ftok()函数被用来生成一个key值，key值可以被msgget()函数、shmget()函数、semget()函数使用，参数pathname（路径名必须存在且可以自定义）与proj_id的低八位（可以自定义）共同产生一个key值：

//为队列随机附加key，pathename为路径，id号可随意（1-255）
key_t ftok(const char *pathname, int proj_id);
1
2

3.2 用户消息缓冲区

无论发送进程还是接收进程，都需要在进程空间中用消息缓冲区来暂存消息。该消息缓冲区的结构定义如下：

struct msgbuf{
	long int msgtype;		//消息类型
	anytype data;			//要发送的数据，可以为任意类型
};
1
2
3
4

通过msgtype区分数据类型，同过判断msgtype，是否为需要接收的数据；
data为存放消息的正文。

3.3 msgget函数

#include <sys/msg.h>
//创建消息队列，返回值为该队列号
int msgget(key_t key, int msgflg); 
1
2
3

功能：创建一个消息队列或获取一个已经存在的消息队列。

参数说明：

• key：传入参数，消息队列的键值，通常为一个整数，若键值为IPC_PRIVATE，将会创建一个只能被创建消息队列的进程读写的消息队列；
• msgflg：参数msgflg用来设置标志位属性，可以指定IPC_CREAT，表示如果消息队列不存在，则自动创建。也可以指定IPC_EXCL，表示如果队列已存在，则返回错误码EEXIST。同时msgflg必须还要指定权限，这一点与文件的mode权限一样。类似于open函数中标志位的功能，用于设置消息队列的创建方式或权限，通常是一个9位的权限与如下值进行位操作后获得：
  – msgflg = mask | IPC_CREAT，若内核中不存在指定消息队列，则它会被创建；若已存在，则获取该消息队列；
  – msgflg = mask | IPC_CREAT | IPC_EXCL时，若消息队列不存在，则它会被创建；若已存在，则msgget函数调用失败，返回-1，并设置errno为EEXIST。

返回值说明：

• 成功：返回消息队列的标识符；
• 不成功：返回-1并设置errno。

一般情况下可以进行如下设置：

int msgget(key, IPC_CREAT|IPC_EXCL|O664);
1

如果消息队列不存在，则自动创建；
如果消息队列存在，则返回错误码EEXIST，表示文件已存在，不需再创建，此时，函数的功能可以认为是创建并打开一个消息队列。
如果队列已存在，只需打开，则函数参数可以设置如下：

int msgget(key, 0664);
1

3.4 msgsnd函数

#include <sys/msg.h>

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);//发送消息
1
2
3

功能：向指定消息队列发送一个消息；如果msgflg = 0，调用函数的进程会被挂起，直到消息写入消息队列为止。

参数说明：

• msqid：消息队列标识符，即msgget函数调用成功后的返回值；
• msgp：指向消息缓冲区的指针；
• msgsz：消息中数据的长度，这个长度不包括消息类型（长整型成员变量）的长度；
• msgflg：标志位，可以设置为0或IPC_NOWAIT。

返回值说明：

• 成功：返回消息队列的标识符；
• 不成功：若消息队列已满或系统中消息数量达到上限，返回-1并设置errno。

3.5 msgrcv函数

#include <sys/msg.h>

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);//接受信息，msgtyp需要指明接受的数据type
1
2
3

功能：从消息队列中读取消息，被读取的消息会从消息列表中移除。

参数说明：

• msqid：消息队列标识符，即msgget函数调用成功后的返回值；
• msgp：指向所读取消息的结构体指针；
• msgsz：消息中数据的长度，这个长度不包括长整型成员变量的长度；
• msgtyp：从消息队列中读取的消息类型：
  – msgtyp = 0：获取队列中的第一个可用消息；
  – msgtyp > 0：获取队列中与该值类型相同的第一个消息；
  – msgtyp < 0：获取队列中消息类型小于或等于其绝对值的第一个消息。
• msgflg：标志位：
  – msgflg = 0：进程将阻塞等待消息的读取；
  – msgflg = IPC_NOWAIT：进程未读取到指定消息时将立刻返回-1。

返回值说明：

• 成功：返回消息队列的标识符；
• 不成功：返回-1并设置errno。

3.5 msgctl函数

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

int msgctl(int msqid, int cmd, struct msqid_ds *buf);//对指定消息队列进行控制
//内核为每个消息队列维护一个msqid_ds结构，用于消息队列的管理
struct msqid_ds {
	struct ipc_perm msg_perm;	//所有者和权限标识
	time_t msg_stime;    		//最后一次发送消息的时间
	time_t msg_rtime;    		//最后一次接收消息的时间
	time_t msg_ctime;			//最后改变的时间
	unsigned long __msg_cbytes;	//队列中当前数据字节数
	msgqnum_t msg_qnum;			//队列中当前消息数
	msglen_t msg_qbytes;		//队列中允许的最大字节数
	pid_t msg_lspid;			//最后发送消息的进程pid
	pid_t msg_lrpid;			//最后接收消息的进程pid
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

功能：对指定消息队列进行控制，如删除、获取属性等。

参数说明：

• msqid：消息队列标识符，即msgget函数调用成功后的返回值；
• cmd：消息队列的处理命令：
  – cmd = IPC_RMID：从系统内核中删除指定命令，使用命令ipcrm -q id可实现同样的功能；
  – cmd = IPC_SET：若进程有权限，将内核管理的消息队列的当前属性值设置为参数buf各成员的值；
  – cmd = IPC_STAT：将内核所管理的消息队列的当前属性值复制给参数buf。
• buf：一个缓冲区，用于传递属性值给指定消息队列或从指定消息队列中获取属性值，其功能视cmd而定。

返回值说明：

• 成功：返回消息队列的标识符；
• 不成功：返回-1并设置errno。

【案例1】使用消息队列实现不同进程间的通信。

• msgsend.c：消息发送端
• msgrcv.c：消息接收端

msgrcv.c
#include <stdio.h>
#include <stdlib.h>
#include <sys/msg.h>
#define MAX_TEXT 512
struct my_msg_st{
	long int my_msg_type;
	char anytext[MAX_TEXT];
};
int main(){
	int tempIdx = 1;
	int tempMsgid;
	struct my_msg_st tempData;
	long int tempMsgToRcv = 0;
	//rcv msg
	tempMsgid = msgget((key_t)1000, 0664 | IPC_CREAT);//获取消息队列
	if (tempMsgid == -1){
		perror("msgget err");
		exit(-1);
	}//of if
	while (tempIdx < 5){
		//接收消息
		if (msgrcv(tempMsgid, (void*)&tempData, BUFSIZ, tempMsgToRcv, 0) == -1){
			perror("msgrcv err");
			exit(-1);
		}//of if
		//打印消息
		printf("msg type:%ld\n", tempData.my_msg_type);
		printf("msg content is:%s", tempData.anytext);
		tempIdx ++;
	}//of while
	//删除消息队列
	if (msgctl(tempMsgid, IPC_RMID, 0) == -1){
		perror("msgctl err");
		exit(-1);
	}//of if
	exit(0);
}//of main
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

msgsend.c
#include <stdio.h>
#include <stdlib.h>
#include <sys/msg.h>
#include <string.h>
#define MAX_TEXT 512
//消息结构体
struct my_msg_st{
	long int my_msg_type;       				//消息类型
	char anytext[MAX_TEXT];     				//消息数据
};
int main() {
	int tempIdx = 1;
	int tempMsgid;
	struct my_msg_st tempData;
	char tempBuf[BUFSIZ];						//设置缓存变量
	tempMsgid = msgget((key_t)1000, 0664 | IPC_CREAT);//创建消息队列
	if (tempMsgid == -1){
		perror("msgget err");
		exit(-1);
	}//of if
	while (tempIdx < 5){							//发送消息
		printf("enter some text:");
		fgets(tempBuf, BUFSIZ, stdin);
		tempData.my_msg_type = rand() % 3 + 1;        	//随机获取消息类型
		strcpy(tempData.anytext, tempBuf);
		//发送消息
		if (msgsnd(tempMsgid, (void*)&tempData, sizeof(tempData), 0) == -1){
			perror("msgsnd err");
			exit(-1);
		}//of if
		tempIdx ++;
	}//of while
	return 0;
}//of main
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

• BUFSIZ：Linux系统定义在stdio.h的宏，表示默认缓冲区大小；

【案例2】消息队列实验：终端2运行msg_b.c输入“hello”，则终端1输出“hello”；终端1运行msg_a.c输入“world”，则终端2输出“world”；当在终端2 输入“quit”时，程序全部退出，并且消息队列删除。

#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <errno.h>
#include <signal.h>

#define N 128
#define SIZE sizeof(struct msgbuf) - sizeof(long)
#define TYPE1 100
#define TYPE2 200

struct msgbuf{
	long mtype;
	char buf[N];
};
int main(int argc, const char *argv[]){
	key_t tempKey;
	/*创建key值*/
	if((tempKey= ftok(".", 'a')) < 0){
		perror("ftok error");
		return -1;
	}//of if
	/*创建或打开消息队列*/
	int tempMsqid;
	struct msgbuf tempMsgSnd, tempMsgRcv;
	if((tempMsqid = msgget(tempKey, IPC_CREAT|IPC_EXCL|0664)) < 0){
		if(errno != EEXIST){
			perror("msgget error");
			return -1;
		}else{
			/*如果消息队列已存在，则打开消息队列*/
			tempMsqid = msgget(tempKey, 0664);
		}//of if
	}//of if

	pid_t tempPid;

	tempPid = fork();

	if(tempPid < 0){
		perror("fork error");
		return -1;
	}else if(tempPid == 0){//子进程
		while(1){
			tempMsgSnd.mtype = TYPE1;
			fgets(tempMsgSnd.buf, N, stdin);
			tempMsgSnd.buf[strlen(tempMsgSnd.buf) - 1] = '\0';

			msgsnd(tempMsqid, &tempMsgSnd, SIZE, 0);

			if(strncmp(tempMsgSnd.buf, "quit", 4) == 0){
				kill(getppid(), SIGKILL);
				break;
			}//of if
		}//of while
	}else{//父进程
		while(1){
			msgrcv(tempMsqid, &tempMsgRcv, SIZE, TYPE2, 0);

			if(strncmp(tempMsgRcv.buf, "quit", 4) == 0){
				kill(tempPid, SIGKILL);
				goto err;
			}//of if

			printf("msg_b:%s\n", tempMsgRcv.buf);
		}//of while
	}//of if
	return 0;
err:
	msgctl(tempMsqid, IPC_RMID, NULL);
}//of main

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

3.6 键值和标识符

• 键值（ID）：ID是msgget函数的返回值，一个非负整数，属于进程间通信的内部名，用来确保使用同一个消息队列。内部名即在进程内部使用，是消息队列在进程级别的唯一标识，这样的标识方法是不能支持进程间通信的。

• 标识符(key)： key是实现进程与消息队列关联的关键，属于进程间通信的外部名，是消息队列在内存级别的唯一标识。当多个进程，针对同一个key调用msgget函数，这些进程得到的ID其实是标识了同一个进程间通信的结构。多个进程间就可以通过这个进程间通信的结构进行通信。