3. 消息队列
3.1 System V IPC
(1)System V IPC概述
①Unix系统存在信号、管道和命名管道等基本进程间通信机机制。
②System V引入了三种高级进程间的通信机制:消息队列、共享内享和信号量。
③IPC对象(消息队列、共享内存和信号量)存在于内核中而不是文件系统中,由用户控制释放(用户管理ipc对象的生命周期),不像管道的释放由内核控制。
④IPC对象通过标识来引用和访问,所有IPC对象在内核空间中有唯一性标识ID,在用户空间中的唯一性标识称为key。
⑤Linux IPC继承自System V IPC
(2)System V IPC对象的访问
①IPC对象是全局对象:可用ipcs、picrm等命令查看或删除。
②每个IPC对象都由get函数创建:msgget、shmget、semget。调用get函数时必须指定关键字key。
(3)IPC对象的权限和所有者结构体
3.2 消息队列
(1)消息队列简介
①消息队列是内核中的一个链表;
②用户进程将数据传输到内核后,内核重新添加一些如用户ID、组ID、读写进程的ID和优先级等相关信息后并封装成一个数据包称为消息。
③允许一个或多个进程往消息队列中写消息和读消息,但一个消息只能被一个进程读取,读取完毕后就自动删除。
④消息队列具有一定的FIFO的特性,消息可以按照顺序发送到队列中,也可以几种不同的方式从队列中读取。每一个消息队列在内核中用一个唯一的IPC标识ID表示。
⑤消息队列的实现包括创建和打开队列、发送消息、读取消息和控制消息队列等四种操作。
(2)消息队列结构(msqid_ds)
(3)打开和创建消息队列
头文件 |
#include <sys/msg.h> |
函数 |
int msgget(key_t key, int flag); |
参数 |
key: 用户指定的消息队列键值 flag:IPC_CREAT、IPC_EXCL等权限组合 |
功能 |
打开或创建消息队列 |
返回值 |
成功返回内核中消息队列的标识ID,出错返回-1 |
备注 |
①若创建消息队列,key可指定键值,也可将之设置为IPC_PRIVATE。 ②若打开进行查询,则key不能为0,必须是一个非零的值,否则查询不到。 |
(4)消息队列的控制
头文件 |
#include <sys/msg.h> |
函数 |
int msgctl(int msgqid, int cmd, struct msqid_ds* buf); |
参数 |
msgqid:消息队列ID buf:消息队列属性指针 cmd: (1)IPC_STAT:获取消息队列的属性,取此队列的msqid_ds结构,并将其存放在buf指定的结构体中。 (2)IPC_SET:设置属性,按由buf指向的结构中的值,设置与此队列相关的结构中的字段。 (3)IPC_RMID:删除队列,从系统中删除该消息队列以及仍在该队列上的所有数据。 |
功能 |
消息队列的控制 |
返回值 |
成功返回0,出错返回-1 |
(5)发送消息
头文件 |
#include <sys/msg.h> |
函数 |
int msgsnd(int msgqid, const void* ptr, size_t nbytes, int flag); |
参数 |
msqid:消息队列ID ptr:用户自定义的结构体,但第一个成员必须是mtype。 struct mymesg{ long mtype; //positive message type,消息的类型,只能是大于0的整数。 char mtext[512]; //message data, of length nbytes,消息数据本身 } nbytes: 指定消息的大小,不包括mtype的大小。 flag: (1)0,阻塞。(2)IPC_NOWAIT:类似于文件I/O的非阻塞I/O标志。 (3)如消息队列己满(或队列中的消息总数等于系统限制值,或队列的字节总数等于系统限制值),则指定的IPC_NOWAIT使用msgsnd立即出错返回EAGAIN。如果指定为0,则进程:A.阻塞直到空间可以容纳要发送的消息。B.或从系统中删除此队列;C.或捕捉到一个信号,并从信号处理程序返回。 |
功能 |
发送消息 |
返回值 |
成功返回0,出错返回-1 |
备注 |
在linux中,消息的最大长度是4056个字节,其中包括mtype,它占有4个字节。 |
(6)接收消息
头文件 |
#include <sys/msg.h> |
函数 |
int msgrcv(int msgqid, void* ptr, size_t nbytes, int flag); |
参数 |
msqid:消息队列ID ptr:指向存放消息的缓存 nbytes:消息缓存的大小,不包括mtype的大小。计算方式: nbytes = sizeof(struct mymesg) – sizeof(long); type:消息类型 (1)type == 0:获取消息队列中的第一个消息 (2)type > 0: 获取消息队列中类型为type的第一个消息 (3)type < 0: 获取消息队列中小于或等于type绝对值的消息(类型最小的) flag:0或IPC_NOWAIT |
功能 |
接收消息 |
返回值 |
成功返回消息的数据部分长度,出错返回-1 |
【编程实验】消息队列
//msq_snd.c //发送进程(要单独编译成一个可执行文件)
#include <stdio.h> #include <stdlib.h> #include <sys/msg.h> //消息结构体 typedef struct{ long type; //消息类型 int start; //消息数据本身(包括start和end) int end; }MSG; /*往消息队列中发送消息*/ int main(int argc, char* argv[]) { //从命令行中转入消息的key if(argc < 2){ printf("usage: %s key\n", argv[0]); exit(1); } key_t key = atoi(argv[1]);//key由用户指定,如10 //key_t key = IPC_PRIVATE; //key由系统指定 //key_t key = ftok(argv[0], 0);//要据文件名和第2个参数生成key printf("key: %d\n", key); //创建或获取消息队列 int msq_id; //IPC_CREAT | IPC_EXCL:不存在则创建,存在时返回错误!(保证对象总是新的) //而不是打开己有的对象 if((msq_id = msgget(key, IPC_CREAT | IPC_EXCL | 0777)) < 0){ perror("msgget error"); } printf("msq id: %d\n", msq_id); //定义要发送的消息 MSG mArr[5] ={ {4, 4, 100}, {2, 2, 200}, {1, 1, 100}, {6, 6, 600},//注意两条type都等于6的消息。 {6, 60, 6000} }; //发送消息到消息队列 int i = 0; for(; i<5; i++){ if(msgsnd(msq_id, &mArr[i], sizeof(MSG)-sizeof(long), IPC_NOWAIT) < 0){ perror("msgsnd error"); } } //发送后获得消息队列中消息的总数 struct msqid_ds ds={0}; if(msgctl(msq_id, IPC_STAT, &ds) < 0){ perror("msgctl error"); } printf("msg total: %d\n", ds.msg_qnum); return 0; } /*输出结果: [root@localhost 11.IPC]# bin/msq_snd 10 //其中的key(10)由用户指定 key: 10 msq id: 65536 msg total: 5 [root@localhost 11.IPC]# ipcs -q //查看消息队列 ------ Message Queues -------- //可以看出有5种消息 key msqid owner perms used-bytes messages 0x0000000a 65536 root 777 40 5 [root@localhost 11.IPC]# ipcrm -q 65536 //删除指定的消息队列 [root@localhost 11.IPC]# ipcs -q ------ Message Queues -------- key msqid owner perms used-bytes messages */
//msg_rcv.c //接收进程(要单独编译成一个可执行文件)
#include <sys/msg.h> #include <stdio.h> #include <stdlib.h> typedef struct{ long type; int start; int end; }MSG; int main(int argc, char* argv[]) { if(argc < 3){ printf("usage: %s key type\n", argv[0]); exit(1); } key_t key = atoi(argv[1]); long type = atoi(argv[2]); //获得指定的消息队列 int msq_id; if((msq_id = msgget(key, 0777)) < 0){ perror("msgget error"); } printf("msg id: %d\n", msq_id); MSG m; if(msgrcv(msq_id, &m, sizeof(MSG)-sizeof(long), type, IPC_NOWAIT) < 0){ perror("msgrcv error"); }else{ printf("type: %d start: %d end: %d\n", m.type, m.start, m.end); } return 0; } /*输出结果: [root@localhost 11.IPC]# bin/msq_snd 10 //send进程,先创建消息队列 key: 10 msq id: 262144 msg total: 5 [root@localhost 11.IPC]# bin/msq_rcv 10 1 //receive进程,从key队列中获得type为1的消息 msg id: 262144 type: 1 start: 1 end: 100 [root@localhost 11.IPC]# bin/msq_rcv 10 2 //获取type为2的消息 msg id: 262144 type: 2 start: 2 end: 200 [root@localhost 11.IPC]# bin/msq_rcv 10 3 //获取type为3的消息(不存在的消息) msg id: 262144 msgrcv error: No message of desired type //获取type为4的消息 [root@localhost 11.IPC]# bin/msq_rcv 10 4 msg id: 262144 type: 4 start: 4 end: 100 [root@localhost 11.IPC]# bin/msq_rcv 10 6 //获取类型为6的第1条消息(先进先出) msg id: 262144 type: 6 start: 6 end: 600 [root@localhost 11.IPC]# bin/msq_rcv 10 6 //获取类型为6的第2条消息 msg id: 262144 type: 6 start: 60 end: 6000 [root@localhost 11.IPC]# ipcs -q ------ Message Queues -------- key msqid owner perms used-bytes messages 0x0000000a 262144 root 777 0 0 [root@localhost 11.IPC]# ipcrm -q 262144 [root@localhost 11.IPC]# ipcs -q ------ Message Queues -------- key msqid owner perms used-bytes messages */