进程间通讯(五).message queue（1）

前言

UNIX/Linux 是多任务的操作系统，通过多个进程分别处理不同事务来实现，如果多个进程要进行协同工作或者争用同一个资源时，互相之间的通讯就很有必要了

进程间通信，Inter process communication，简称 IPC，在 UNIX/Linux 下主要有以下几种方式:

• 无名管道 ( pipe )
• 有名管道 ( fifo )
• 信号 ( signal )
• 信号量 ( semaphore )
• 消息队列 ( message queues )
• 共享内存 ( shared memory )
• 套接字 ( socket )

这里分享一下我在学习进程通讯过程中的笔记和心得

---

概要

---

消息队列

系统层面的消息队列是消息的链接表，存储在内核中，由消息队列标识符标识

一个消息可以看成一个记录，具有特定的格式以及特定的类别

对消息队列有写权限的进程可以向消息队列中按照一定的规则添加新消息；对消息队列有读权限的进程则可以从消息队列中读走消息。消息队列是随内核持续的。也就是说进程的退出，如果不自主去释放资源，消息队列是会悄无声息的存在的。所以较管道来说，消息队列的生命周期更加持久。消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。每个数据块都被认为是有一个类型，接收者进程接收的数据块可以有不同的类型值。我们可以通过发送消息 来避免命名管道的同步和阻塞问题。消息队列与管道不同的是，消息队列是基于消息的， 而管道是基于字节流的，且消息队列的读取不一定是先入先出。消息队列与命名管道有一 样的不足，就是每个消息的最大长度是有上限的（MSGMAX），每个消息队列的总的字节数是有上限的（MSGMNB），系统上消息队列的总数也有一个上限（MSGMNI）

在某个进程往一个队列写入消息之前，并不需要另外某个进程在该队列上等待消息的到达

pipe 和 FIFO 最后一次关闭发生时，仍在该管道或FIFO上的数据将被丢弃，消息队列，除非内核自举或显式删除，否则其一直存在

管道和FIFO都是随进程持续的，XSI IPC(消息队列、信号量、共享内存)都是随内核持续的

消息队列的链表结构类型于下：

代码语言：javascript复制

/* Obsolete, used only for backwards compatibility and libc5 compiles */
struct msqid_ds {
        struct ipc_perm msg_perm;
        struct msg *msg_first;          /* first message on queue,unused  */
        struct msg *msg_last;           /* last message in queue,unused */
        __kernel_time_t msg_stime;      /* last msgsnd time */
        __kernel_time_t msg_rtime;      /* last msgrcv time */
        __kernel_time_t msg_ctime;      /* last change time */
        unsigned long  msg_lcbytes;     /* Reuse junk fields for 32 bit */
        unsigned long  msg_lqbytes;     /* ditto */
        unsigned short msg_cbytes;      /* current number of bytes on queue */
        unsigned short msg_qnum;        /* number of messages in queue */
        unsigned short msg_qbytes;      /* max number of bytes on queue */
        __kernel_ipc_pid_t msg_lspid;   /* pid of last msgsnd */
        __kernel_ipc_pid_t msg_lrpid;   /* last receive pid */
};

与系统级别相对应的还有应用层面的MQ，(使用一样的思想进行应用解耦)，开源产品有rabbitmq ：MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求

---

系统限制

系统层面有一些内核参数限制了消息队列的大小

代码语言：javascript复制

root@ubuntu:~# sysctl -a 2> /dev/null  | grep msg 
kernel.msgmax = 8192
kernel.msgmni = 1730
kernel.msgmnb = 16384
kernel.auto_msgmni = 1
fs.mqueue.msg_max = 10
fs.mqueue.msgsize_max = 8192
root@ubuntu:~# vim /etc/sysctl.conf 
root@ubuntu:~# 

msgmni

msgmni 定义了系统范围内的消息队列上限。与信号量一样，消息队列也拥有一个相关的标识符。在系统初始化阶段里，内核创建一个指向消息队列标识符结构的指针数组。该数组的项数由 msgmni确定。对于每个消息队列，Linux 内核为标识符分配44B，为消息队列数据结构分配 96B。为了获得更多的消息队列资源，可以动态增加 msgmni 取值。和信号量一样，消息队列标识符的最大数目也受限于IPCMNI。msgmni的默认上限为 16B，这可能不足以保证一些大型数据库应用平滑地运行。如果在系统上要运行数据库应用的话，推荐默认上限值是 128B

msgmax

msgmax 限制进程可以发送的消息长度。该参数由 Msgsnd()函数加以应用。如果待发送消息的长度超过该值，则返回一个错误。该参数可以在运行时调整

---

msgmnb

msgmnb 确定一个消息队列的容量。该参数的取值存储在消息队列标识符结构的某个域中，用于确定是否存在着对新消息进行排队的空间。msgmnb 值可以动态修改，默认为16384。修改其取值会影响到所有新的消息队列的容量。用户可以通过 Msgctl()系统调用来增加现有消息队列的容量

Tip: /etc/sysctl.conf 中可以进行内核配置

---

代码示例

要求

• 1.A、B两个进程（非亲缘关系），A进程往消息队列中写入任意字符串（以START开头则为有效的），B进程读取
• 2.直到收到“quit”才退出

代码示例

msgqueA.c

代码语言：javascript复制

#include <stdio.h>
#include <sys/msg.h> //key_t,ftok,msgget,msgsnd,IPC_CREAT 等相关声明都在这里面定义
#include <unistd.h> //getpid 的函数声明在这个头文件里
#include <string.h> //strncmp,strcmp,strlen 的函数声明在这个头文件里
#define BUFSZ 1024


typedef struct message //此结构体用于存放消息，从中可以看到消息的两个字段
{
  long msg_type; //消息类型，以整型值进行标示
  char msg_text[BUFSZ]; //消息内容
}MSG; //取了一个别名

int main() 
{
  int res=-1,qid=0;
  key_t key=IPC_PRIVATE; //IPC_PRIVATE 就是0  
  int len=0; //赋初值是一个好习惯
  MSG msg;
  
  if(-1 == (key=ftok("/",18))) // 通过ftok获取一个key，两个进程可以通过这个key来获取队列信息
  {
    perror("ftok");
    return res;
  }

  if(-1==(qid=msgget(key,IPC_CREAT|0600))) //创建一个消息队列，将id存到qid中
  {
    perror("msgget");
    return res;
  }
  
  printf("open queue %dn",qid); //将qid进行显示
  
  while(1)
  {
    puts("please enter the message to queue:n(message start with 'START' will be valid,'quit' to exit)"); 
    if(NULL == (fgets(msg.msg_text,BUFSZ,stdin))) //从标准输入中获取信息放到 msg.msg_text 中
    {
      perror("fgets");
      return res;
    }
    msg.msg_type=getpid(); //将消息的类型设置为本进程的ID
    if( 0== strncmp(msg.msg_text,"START",5) || 0== strcmp(msg.msg_text,"quitn")) //如果内容是以 START 开头，并且不是 quit，就将这条信息发送
    {
      len=strlen(msg.msg_text);
      if (0 > msgsnd(qid,&msg,len,0)) //发送信息
      {
	perror("msgsnd");
	return res;
      }
    }
    if ( 0 == strcmp(msg.msg_text,"quitn") ) break; //如果是quit 就进行退出
  }    
  res=0;
  return res;
}

消息队列CMQ版 kafka bash linux tcp/ip

0 人点赞