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消息队列属性

1.4 函数

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

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

1.4.1 创建消息队列

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

• 参数

• 标志

• 权限

• 返回值

• 示例
  创建一个名字为/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)

• 参数

• 返回值

• 示例
  删除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)

• 参数

• 返回值

• 示例

#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)

• 参数

• 返回值

• 示例
  关闭消息队列

#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)

• 参数

• 示例

#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)

• 参数

• 返回值

• 示例

#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);

• 参数

• 返回值

• 示例

#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)

• 参数

• 返回值

• 示例

#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属性只能被获取而不能被设置。