【Linux】进程间通信之消息队列

简介: 【Linux】进程间通信之消息队列

一、消息队列的原理

消息队列是Linux的一种通信机制,这种通信机制传递的数据具有某种结构,而不是简单的字节流

Linux内核我们可以创建一个队列结构,然后我们可以将我们需要发送和读取的数据插入这个队列里面,多个不同的进程可以通过相同的key值找到相同的队列。

对于消息队列来说:无论发送消息的进程还是接收消息的进程,都需要在进程空间中用消息缓冲区来暂存消息,然后向消息队列写入或读取数据时也按照结构体的方式来进行写入和读取的!对于进程来说消息缓冲区的结构定义一般如下:

由于结构体中有一个mtype类型,这个字段可以帮助我们区分是哪一个进程写入的,消息队列里面可以让多个不同的进程写入数据,多个不同的进程读取数据,因此消息队列是全双工通信,可读可写

消息队列的内核结构

  • 消息队列的本质其实是一个内核提供的链表,内核基于这个链表,实现了一个数据结构。
  • 向消息队列中写数据,实际上是向这个数据结构中插入一个新结点;从消息队列读数据,实际上是从这个数据结构中删除一个结点。
  • 和管道一样,每个消息的最大长度是有上限的(MSGMAX),每个消息队列的总字节数也是有上限的(MSGMNB),系统上的消息队列总数也是有上限的(MSGMNI)
  • 消息队列是一个全双工通信,可读可写
  • 消息队列的生命周期是随内核的,即进程退出以后消息队列不会消失!

二、消息队列的使用

和共享内存一样,消息队列的使用也涉及很多的系统调用,而且它们的调用接口都是类似的,下面我们开始进行讲解。

1、msgget函数

此函数用于帮我们创建消息队列,创建完毕以后会给我们返回一个消息队列的标识符。

  • 参数
  1. 与共享内存一样,是一个key值,可以通过ftok函数进行获取
  2. 是一个标志位,主要有三个标志:IPC_CREATIPC_EXCLmode_flags 其含义与共享内存一致。
  • 返回值:调用成功,返回一个和参数key相关联的消息队列的标识符,调用失败就返回 -1,错误码被设置。

2、msgctl函数

此函数的功能很强大,里面有许多的标志位,可以完成许多不同的工作,这个函数主要用来控制消息队列。

  • 参数
  1. 来自于msgget函数得到的消息队列标识符。
  2. 是标志位,里面有许多标志,我们经常使用的是这两个: IPC_RMIDIPC_STAT
  3. 一个struct msqid_ds类型的指针,在标志位中设置了IPC_STAT,指针所指向的变量里面就能拿到相关的内核信息,如果不关心内核信息可以设置为nullptr
  • 返回值:一般来说,成功返回是0,错误返回结果是-1

3、msgsnd函数

此函数可以将我们要通信的消息放入消息队列里面,类似与Linux文件操作中的write函数。

  • 参数
  1. 来自于msgget函数得到的消息队列标识符。
  2. 要写入的消息的指针,由于消息队列支持多个进程进行写入,在向消息队列里面写数据时,用户自己要组织一个结构体,然后将这个结构体对象当成一条消息进行写入。实际中对于此参数的结构体常常这样定义:

  3. 发送的消息正文的字节数,注意这里的是指正文内容mtext里面数据的字节数。
  4. 标志位,IPC_NOWAIT消息没有发送完成函数也会立即返回,0:直到发送完成函数才返回。
  • 返回值:成功返回是0,错误返回结果是-1

4、msgrcv函数

这个函数的作用可以帮助我们从消息队列里面取出数据,类似与Linux文件操作中的read函数。

  • 参数
  1. 来自于msgget函数得到的消息队列标识符。
  2. 读取到的数据要放到哪里,这里还是填我们自定义的结构体对象。
  3. 要读取的正文字节数
  4. 消息队列里面的消息的区分类型mtype,选择你想要读取的类型的数据
  5. 标准位,IPC_NOWAIT,非阻塞等待,若没有消息,进程会立即返回 -1

    0,阻塞等待。
  • 返回值
    如果读取成功就返回读取到的字节数,如果读取失败就返回-1,错误码被设置。

5、代码实例

command.hpp

#ifndef __COMMAND_HPP
#define __COMMAND_HPP
#include <iostream>
#include <string>
#include <cstdlib>
#include <cstring>
#include <cerrno>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
const std::string pathname = "./";
const int proj_id = 10;
struct msgqbuf
{
    long mtype;      // 该字段用于区分是哪一个进程写的数据 , message type, must be > 0 
    char mtext[];   // 该字段才是正文内容, 这里可以给变长数组,也可以给一个确定大小的数组
};
#endif

msgserver.cpp

#include "command.hpp"
int main()
{
    // 1.生成key值
    key_t key = ftok(pathname.c_str(), proj_id);
    if (key == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
        return -1;
    }
    // 2.创建消息队列
    int msqid = msgget(key, IPC_CREAT | IPC_EXCL | 0664);
    if (msqid == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
        return -1;
    }
    std::cout << "消息队列创建成功" << std::endl;
    // 3.进行进程间通信
    // 读取消息
    msgqbuf* pmb_r = (msgqbuf*)malloc(sizeof(msgqbuf) + sizeof(char) * 25);
    pmb_r->mtype = 1;
    int readnum = msgrcv(msqid, pmb_r, sizeof(char) * 25, pmb_r->mtype, 0);
    if (readnum == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
    }
    else if (readnum > 0)
    {
        std::cout << "读取到了" << readnum << "字节" << std::endl;
        std::cout << "消息的类型: " << pmb_r->mtype << " 消息的内容: " << pmb_r->mtext << std::endl;
    }
    // 4.删除消息队列
    int err = msgctl(msqid, IPC_RMID, nullptr);
    if (err == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
    }
    free(pmb_r);
    return 0;
}

msgclient.cpp

#include "command.hpp"
int main()
{
    // 1.生成key值
    key_t key = ftok(pathname.c_str(), proj_id);
    if (key == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
        return -1;
    }
    // 2.创建消息队列
    int msqid = msgget(key, IPC_CREAT);
    if (msqid == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
        return -1;
    }
    // 3.进行进程间通信
    // 3.1向消息队列发送消息
    msgqbuf* pmb_w = (msgqbuf*)malloc(sizeof(msgqbuf) + sizeof(char) * 25);
    if (!pmb_w)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
    }
    // 设置消息的类型与数据
    pmb_w->mtype = 1;
    std::string s = "hello linux\n";
    int i = 0;
    for (auto& e : s)
    {
        pmb_w->mtext[i++] = e;
    }
    // 发送消息
    int err = msgsnd(msqid, pmb_w, strlen(pmb_w->mtext), 0);
    if (err == -1)
    {
        std::cerr << "错误码 : " << errno << " " << strerror(errno) << std::endl;
    }
    free(pmb_w);
    return 0;
}

要先运行msgserver创建消息队列,然后运行msgclient进行通信。

运行结果:

相关文章
|
2天前
|
缓存 监控 Linux
linux进程管理万字详解!!!
本文档介绍了Linux系统中进程管理、系统负载监控、内存监控和磁盘监控的基本概念和常用命令。主要内容包括: 1. **进程管理**: - **进程介绍**:程序与进程的关系、进程的生命周期、查看进程号和父进程号的方法。 - **进程监控命令**:`ps`、`pstree`、`pidof`、`top`、`htop`、`lsof`等命令的使用方法和案例。 - **进程管理命令**:控制信号、`kill`、`pkill`、`killall`、前台和后台运行、`screen`、`nohup`等命令的使用方法和案例。
19 4
linux进程管理万字详解!!!
|
1天前
|
算法 Linux 定位技术
Linux内核中的进程调度算法解析####
【10月更文挑战第29天】 本文深入剖析了Linux操作系统的心脏——内核中至关重要的组成部分之一,即进程调度机制。不同于传统的摘要概述,我们将通过一段引人入胜的故事线来揭开进程调度算法的神秘面纱,展现其背后的精妙设计与复杂逻辑,让读者仿佛跟随一位虚拟的“进程侦探”,一步步探索Linux如何高效、公平地管理众多进程,确保系统资源的最优分配与利用。 ####
15 4
|
2天前
|
消息中间件 存储 供应链
进程间通信方式-----消息队列通信
【10月更文挑战第29天】消息队列通信是一种强大而灵活的进程间通信机制,它通过异步通信、解耦和缓冲等特性,为分布式系统和多进程应用提供了高效的通信方式。在实际应用中,需要根据具体的需求和场景,合理地选择和使用消息队列,以充分发挥其优势,同时注意其可能带来的复杂性和性能开销等问题。
|
2天前
|
缓存 负载均衡 算法
Linux内核中的进程调度算法解析####
本文深入探讨了Linux操作系统核心组件之一——进程调度器,着重分析了其采用的CFS(完全公平调度器)算法。不同于传统摘要对研究背景、方法、结果和结论的概述,本文摘要将直接揭示CFS算法的核心优势及其在现代多核处理器环境下如何实现高效、公平的资源分配,同时简要提及该算法如何优化系统响应时间和吞吐量,为读者快速构建对Linux进程调度机制的认知框架。 ####
|
4天前
|
消息中间件 存储 Linux
|
10天前
|
运维 Linux
Linux查找占用的端口,并杀死进程的简单方法
通过上述步骤和命令,您能够迅速识别并根据实际情况管理Linux系统中占用特定端口的进程。为了获得更全面的服务器管理技巧和解决方案,提供了丰富的资源和专业服务,是您提升运维技能的理想选择。
10 1
|
22天前
|
算法 Linux 调度
深入理解Linux操作系统的进程管理
【10月更文挑战第9天】本文将深入浅出地介绍Linux系统中的进程管理机制,包括进程的概念、状态、调度以及如何在Linux环境下进行进程控制。我们将通过直观的语言和生动的比喻,让读者轻松掌握这一核心概念。文章不仅适合初学者构建基础,也能帮助有经验的用户加深对进程管理的理解。
16 1
|
27天前
|
消息中间件 Linux API
Linux c/c++之IPC进程间通信
这篇文章详细介绍了Linux下C/C++进程间通信(IPC)的三种主要技术:共享内存、消息队列和信号量,包括它们的编程模型、API函数原型、优势与缺点,并通过示例代码展示了它们的创建、使用和管理方法。
23 0
Linux c/c++之IPC进程间通信
|
27天前
|
Linux C++
Linux c/c++进程间通信(1)
这篇文章介绍了Linux下C/C++进程间通信的几种方式,包括普通文件、文件映射虚拟内存、管道通信(FIFO),并提供了示例代码和标准输入输出设备的应用。
19 0
Linux c/c++进程间通信(1)
|
4月前
|
消息中间件 C语言 RocketMQ
消息队列 MQ操作报错合集之出现"Connection reset by peer"的错误,该如何处理
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。