Linux系统编程4：IPC消息队列

0. 消息队列

• 背景
  管道和套接字比较适合两三个进程之间的通信，如果进程成倍增加，管道和套接字的数量也会阶乘级的增加，管理也非常复杂麻烦。于是出现的消息队列

• 比喻

  
  回转寿司

• 优先级
  让列宁同志先走

• 本质
  内核链表

1. POSIX 消息队列

1.1 查看

• POSIX消息队列预览：man mq_overview
• 查看POSIX消息队列：ls /dev/mqueue
• cat /dev/mqueue/PIC名字

1.2 接口

• 头文件:mqueue.h
• 库:librt.so(real time)

1.3 结构体

• struct mq_attr消息队列属性

No.	成员	含义	说明
1	mq_flags	标志	在mq_open时被初始化,在mq_setattr设置,其值为0或者O_NONBLOCK
2	mq_maxmsg	队列的消息个数最大值	只能在mq_open时被初始化
3	mq_msgsize	队列每个消息长度的最大值	只能在mq_open时被初始化
4	mq_curmsgs	当前队列消息个数	在mq_getattr获取

1.4 函数

POSIX 消息队列主要有八个操作

No.	操作	函数
1	创建消息队列	mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr)
2	删除消息队列	int mq_unlink(const char *name)
3	打开消息队列	mqd_t mq_open(const char *name, int oflag)
4	关闭消息队列	int mq_close(mqd_t mqdes)
5	发送消息	int mq_send(mqd_t mqdes, const char *msg_ptr,size_t msg_len, unsigned msg_prio)
6	接收消息	ssize_t mq_receive(mqd_t mqdes, char *msg_ptr,size_t msg_len, unsigned *msg_prio)
7	设置消息队列属性	int mq_setattr(mqd_t mqdes, struct mq_attr *newattr,struct mq_attr *oldattr)
8	获取消息队列属性	int mq_getattr(mqd_t mqdes, struct mq_attr *attr)

*消息队列操作非常相似文件读写操作。

1.4.1 创建消息队列

mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr)

• 参数

No.	参数	含义
1	name	posix IPC名字，格式为/somename
2	oflag	标志
3	mode	权限
4	attr	队列属性,阻塞attr.mq_flag = 0;非阻塞attr.mq_flag = NONBLOCK

• 标志

No.	标志	作用
1	O_CREAT	没有该对象则创建
2	O_EXCL	如果O_CREAT指定，但name不存在，就返回错误
3	O_NONBLOCK	以非阻塞方式打开消息队列
4	O_RDONLY	只读
5	O_RDWR	读写
6	O_WRONLY	只写

• 权限

No.	权限	作用
1	S_IWUSR	用户/属主写
2	S_IRUSR	用户/属主读
3	S_IWGRP	组成员写
4	S_IRGRP	组成员读
5	S_IWOTH	其他用户写
6	S_IROTH	其他用户读

• 返回值

No.	返回值	含义
1	-1	出错
2	其他	消息队列描述符

• 示例
  创建一个名字为/tmp.test的POSIX 消息队列

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open()
#define FILE_MODE (S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH) //0644
int main(){
    struct mq_attr attr;
    attr.mq_maxmsg = 100;
    attr.mq_msgsize = 100;
    mqd_t mqd = mq_open("/tmp.test",O_CREAT,FILE_MODE,&attr);
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
}

注意编译时可能需要加上 -lrt。

• 问题

1. 是否会创建/tmp.test这个文件？
2. 执行命令ll /dev/mqueue会发现什么？
3. 执行命令cat /dev/mqueue/tmp.test会发现什么？
4. Posix IPC名字可以是abc、/abc、/abc.def、/abc/def？
5. 系统限制是否可以修改？

• 注意
  POSIX消息队列的名字所创建的真正路径名和具体的系统实现有关，关于具体POSIX IPC的名字规则可以参考《UNIX 网络编程 卷2：进程间通信》。在Redhat所创建的POSIX消息队列不会在文件系统中创建真正的路径名。且POSIX的名字只能以一个/开头，名字中不能包含其他的/。

1.4.2 删除消息队列

int mq_unlink(const char *name)

• 参数

No.	参数	含义
1	name	posix IPC名字

• 返回值

No.	返回值	含义
1	-1	出错
2	0	成功

• 示例
  删除POSIX消息队列

#include <mqueue.h> // mq_unlink()
int main(){
    mq_unlink("/tmp.test");
}

• 问题

1. 如何验证是否删除？

1.4.3 打开消息队列

mqd_t mq_open(const char *name, int oflag)

• 参数

No.	参数	含义
1	name	posix IPC名字
2	oflag	标志,O_RDONLY只读;O_RDWR读写;O_WRONLY只写

• 返回值

No.	返回值	含义
1	-1	出错
2	其他	描述符

• 示例

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open()
int main(){
    mqd_t mqd = mq_open("/tmp.test",O_RDONLY);
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
    return 0;
}

• 问题

1. 如果消息队列不存在，会发生什么情况？

1.4.4 关闭消息队列

int mq_close(mqd_t mqdes)

• 参数

No.	参数	含义
1	mqdes	消息队列描述符

• 返回值

No.	返回值	含义
1	-1	出错
2	0	成功

• 示例
  关闭消息队列

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open() mq_close()
int main(){
    mqd_t mqd = mq_open("/tmp.test",O_RDONLY);
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
    // ...
    mq_close(mqd);
}

• 注意
  mq_close()和文件的close()类似，关闭后，消息队列并不从系统中删除。一个进程结束，会自动调用关闭打开着的消息队列。

1.4.5 发送消息

int mq_send(mqd_t mqdes, const char *msg_ptr,size_t msg_len, unsigned msg_prio)

• 参数

No.	参数	含义
1	msg_ptr	消息的指针。
2	msg_len	消息长度，不能大于属性值mq_msgsize的值
3	msg_prio	优先级，小于MQ_PRIO_MAX，数值越大，优先级越高

• 示例

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open() mq_send() mq_close()
int main(int argc,char* argv[]){
    mqd_t mqd = mq_open("/tmp.test",O_WRONLY); // 可以设置O_NONBLOCK
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
    const char* msg = "HelloWorld";
    if(-1 == mq_send(mqd,msg,sizeof(msg),1)){
        perror("mq_send error");
        mq_close(mqd);
        return 1;
    }
    mq_close(mqd);
    return 0;
}

• 问题

  1. 如果发送的消息长度超过mq_msgsize会怎样？
  2. 如果发送的消息个数超过mq_maxmsg会怎样？

• 注意
  消息在队列中将按照优先级大小顺序来排列消息
  消息队列已满，mq_send()函数将阻塞,直到有可用空间再次允许放置消息。
  如果O_NONBLOCK被指定，mq_send()那么将不会阻塞，而是返回EAGAIN错误。

1.4.6 接收消息

ssize_t mq_receive(mqd_t mqdes, char *msg_ptr,size_t msg_len, unsigned *msg_prio)

• 参数

No.	参数	含义
1	msg_ptr	消息的指针。
2	msg_len	消息长度，不能大于属性值mq_msgsize的值
3	msg_prio	优先级，消息在队列中将按照优先级大小顺序来排列消息

• 返回值

No.	返回值	含义
1	-1	出错
2	正数	接收到的消息长度

• 示例

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open() mq_receive() mq_close()
#define FILE_MODE (S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH) //0644
int main(int argc,char* argv[]){
    mqd_t mqd = mq_open("/tmp.test",O_RDONLY);
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
    char buf[BUFSIZ];
    unsigned int prio;
    if(-1 == mq_receive(mqd,buf,BUFSIZ,&prio)){
        perror("mq_send error");
        mq_close(mqd);
        return 1;
    }
    printf("msg:%s\nprio:%d\n",buf,prio);
    mq_close(mqd);
    return 0;
}

• 问题

  1. 如果消息队列为空，接受消息会怎样？

• 注意

  1. POSIX消息队列在调用mq_receive()时总是返回队列中最高优先级的最早消息。
  2. 如果队列空，mq_receive()函数将阻塞，直到消息队列中有新的消息。
  3. 如果O_NONBLOCK被指定，mq_receive()那么将不会阻塞，而是返回EAGAIN错误。

1.4.7 设置消息队列属性

int mq_setattr(mqd_t mqdes, struct mq_attr *newattr,struct mq_attr *oldattr);

• 参数

No.	参数	含义
1	mqdes	消息队列描述符
2	newattr	新属性,只能设置mq_flags:0:NONBLOCK
3	oldattr	旧属性

• 返回值

No.	返回值	含义
1	-1	出错
2	0	成功

• 示例

#include <stdio.h> // perror()
#include <string.h> // bzero()
#include <mqueue.h> // mq_open() mq_setattr() mq_close()
int main(){
    mqd_t mqd = mq_open("/tmp.test",O_RDWR);
    if(-1 == mqd){
        perror("mq_open error");
        return 1;
    }
    struct mq_attr new_attr;
    bzero(&new_attr,sizeof(new_attr));
 
    new_attr.mq_flags = O_NONBLOCK;
    struct mq_attr attr;
    if(-1 == mq_setattr(mqd,&new_attr,&attr)){
        perror("mq_setattr error");
        mq_close(mqd);
        return 1;
    }
    printf("flag:%ld,Max msg:%ld,Max msgsize:%ld,Cur msgnun:%ld\n",attr.mq_flags,attr.mq_maxmsg,attr.mq_msgsize,attr.mq_curmsgs);
    mq_close(mqd);
}

1.4.8 获取消息队列属性

int mq_getattr(mqd_t mqdes, struct mq_attr *attr)

• 参数

No.	参数	含义
1	mqdes	消息队列描述符
2	attr	属性

• 返回值

No.	返回值	含义
1	-1	出错
2	0	成功

• 示例

#include <stdio.h> // perror()
#include <mqueue.h> // mq_open() mq_getattr() mq_close()
int main(){
    mqd_t mqd = mq_open("/tmp.test",O_RDONLY);
    if(-1 == mqd){
        perror("mq_open error");
        mq_close(mqd);
        return 1;
    }
    struct mq_attr attr;
    mq_getattr(mqd,&attr);
    printf("flag:%ld,Max msg:%ld,Max msgsize:%ld,Cur msgnun:%ld\n",attr.mq_flags,attr.mq_maxmsg,attr.mq_msgsize,attr.mq_curmsgs);
   mq_close(mqd);
    return 0;
}

• 注意
  1. mq_setattr()可以设置的属性只有mq_flags，用来设置或清除消息队列的非阻塞标志。newattr结构的其他属性被忽略。
  2. mq_maxmsg和mq_msgsize属性只能在创建消息队列时通过mq_open()来设置。
  3. mq_open()只会设置该两个属性，忽略另外两个属性。
  4. mq_curmsgs属性只能被获取而不能被设置。