第11章 进程间通信（2）_消息队列

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
", 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
", 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
", 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
", 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
", 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
", 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
", 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  
 */