进程间通信(IPC)之消息队列

Posted 2023-02-26 Xiao__Tian__

★IPC方法包括管道（PIPE）、消息队列（Message_Queue）、旗语、共用内存（ShareMemory）以及套接字（Socket）。进

程间通信主要包括了管道、系统IPC（包括了消息队列、信号以及共享存储）、套接字（SOCKET）。此文将详细叙述消息队列的相

关内容。

★产生原因：

所谓消息队列，其实就是消息(数据)传输过程中保存的容器。既然有了管道通信的方式，何必又有消息队列呢？

因为根据管道的特性，我们知道其在一定程度上存在或多或少的局限性，首先匿名管道以及命名管道是随进程的，进

程的退出，意味着管道生命周期的结束；其次，管道传送数据时以无格式字节流的形式传送，这有时会给程序的开发

带来不便；再者，担当数据传送媒介的管道，其缓冲区的大小也有较大的限制。所以作为IPC方式下的另一种应用于

进程间通信的消息队列方式便应运而生了。

★什么是消息队列？

  消息队列就是一个消息的链表。可以把消息看作一个记录，具有特定的格式以及特定的优先级。对消息队列有写

权限的进程可以向消息队列中按照一定的规则添加新消息；对消息队列有读权限的进程则可以从消息队列中读走消息

。消息队列是随内核持续的。也就是说进程的退出，如果不自主去释放资源，消息队列是会悄无声息的存在的。所以

较管道来说，消息队列的生命周期更加持久。消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。 

 每个数据块都被认为是有一个类型，接收者进程接收的数据块可以有不同的类型值。我们可以通过发送消息 来避免

命名管道的同步和阻塞问题。消息队列与管道不同的是，消息队列是基于消息的， 而管道是基于字节流的，且消息队

列的读取不一定是先入先出。消息队列与命名管道有一 样的不足，就是每个消息的最大长度是有上限的

（MSGMAX），每个消息队列的总的字节数是有上限的（MSGMNB），系统上消息队列的总数也有一个上限

（MSGMNI）。见下图：

★消息队列的类型：

     目前有POSIX消息队列和系统V消息队列，系统V消息队列目前被大量使用。考虑到程序的可移植性，新开发的

应用程序应尽量使用POSIX消息队列。

系统V消息队列是随内核持续的，只有在内核重起或者显示删除一个消息队列时，该消息队列才会真正被删除

（对比管道，管道的生命周期是随进程的）。因此系统中记录消息队列的数据结构（struct ipc_ids msg_ids）位于

内核中，系统中的所有消息队列都可以在结构msg_ids中找到访问入口。 消息队列就是一个消息的链表。每个消息队

列都有一个队列头，用结构struct msg_queue来描述。队列头中包含了该消息队列的大量信息，包括消息队列键

值、用户ID、组ID、消息队列中消息数目等等，甚至记录了最近对消息队列读写进程的ID。读者可以访问这些信息，

也可以设置其中的某些信息。 

既然消息队列是IPC（Inter Process Communication）中的一种，所以我们先来看看内核为每一个进程间通信对象所维护的数据

结构。

         命令：vim /usr/include/linux/ipc.h

注：可以看到，该结构体中包含了许多关于id方面的信息，key类似于端口号，为目标获取（get）时的标志；

uid为拥有者的id；gid为组用户id；cuid中的c为创建者(creator)；cgid同理；mode为模式，也就是权限；seq

为顺序值。

再来看看关于消息队列中的结构体：

   命令：vim /usr/include/linux/msg.h

注：我们可以看到红色框中的内容正是上面IPC的结构体。而后消息队列私有的数据成员为私有部分，其内容对应的都有注释，就不

一一叙述了。

★相关函数：

1.msgget

注：msgget函数包含了两个参数key和msgflg。参数key类似于端口号，也可以由fork函数生成。参数msgflg中存在两个IPC标

志，即IPC_CREAT和IPC_EXEL。

其中，①IPC_CREAT标志：如果IPC不存在，则创建一个IPC资源，否则打开操作。                                       

  ②IPC_EXCL：只有在共享内存不存在的时候，新的共享内存才建立，否则就产生错误。如果单独使用IPC_CREAT，XXXget()

函数要么返回一个已经存在的共享内存的操作符，要 么返回一个新建的共享内存的标识符。         

  ③如果将IPC_CREAT和IPC_EXCL标志一起使用，XXXget()将返回一个新建的IPC标识符 ；如果该IPC资源已存在，或者返

回-1。IPC_EXEL标志本身并没有太大的意义，但是和IPC_CREAT标志一起使用可以用来保证 所得的对象是新建的，而不是打开已有

的对象。

2.msgrcv和msgsnd：

注：msgrcv从队列中取出消息；msgsnd将数据放到消息队列中；

函数参数：msqid------消息队列的标识码；msgp-------指向消息缓冲区的指针，此位置用来暂时存储发送和接收的消息，是一个

用户可 定义的通用结构，形态如下：struct msgstru     long mtype; //大于0          char mtext[用户指定大⼩小]; ；msgsz------

---消息的大小；msgtyp---------从消息队列内读取的消息形态。如果值为零，则表示消息队列中的所有消息都会被读取；msgflg--

------用来指明核心程序在队列没有数据的情况下所应采取的行动。如果msgflg和常数IPC_NOWAIT合用，则在msgsnd()执行时若

是消息队列已满，则msgsnd()将不会阻塞，而会立即返回-1，如果执行的是msgrcv()，则在消息队列呈空时，不做等待马上返

回-1，并设定错误码为ENOMSG。当msgflg为0时，msgsnd()及msgrcv()在队列呈满或呈空的情形时，采取阻塞等待的处理模式。

3.msgctl：

 

注：该函数设置消息队列的属性。函数参数：msgctl系统调用对msgqid标识的消息队列执行cmd操作，系统定义了3种cmd操作：

IPC_STAT , IPC_SET , IPC_RMID
。

IPC_STAT : 该命令⽤用来获取消息队列对应的 msqid_ds 数据结构，并将其保存到 buf 指 定的地址空间。
       

IPC_SET : 该命令⽤用来设置消息队列的属性，要设置的属性存储在buf中。            

IPC_RMID : 从内核中删除 msqid 标识的消息队列。 

★用消息队列对客户端和服务端间通信的模拟：

1.comm.h

#pragma once  
  
#include<stdio.h>  
#include<errno.h>  
#include<string.h>  
#include<sys/types.h>  
#include<sys/ipc.h>  
#include<sys/msg.h>  
  
#define _PATH_NAME_ "/tmp"  
#define _PROJ_ID_ 0x6666  
#define _SIZE_ 1024  
  
extern int server_type;  
extern int client_type;  
  
  
struct msgbuf  
  
    long mtype;  
    char mtext[_SIZE_];  
;  
  
int creat_msg_queue();  
int get_msg_queue();  
//int creat_msg_queue(int msg_id);  
int send_msg(int msg_id,int send_type,const char* msg);  
int recv_msg(int msg_id,int recv_type,char* msg_out);  
int destroy_queue(int msg_id);  

2.comm.c

#include"comm.h"  
  
int server_type = 1;  
int client_type = 2;  
  
  
static int comm_msg_queue(int flags)  
  
    key_t _key = ftok(_PATH_NAME_,_PROJ_ID_);  
    if(_key < 0)  
      
        printf("%d : %s",errno,strerror(errno));  
        return -1;  
      
  
    int msg_id = msgget(_key,flags);  
    //int msg_id = msgget(_key,IPC_CREAT | IPC_EXCL | 0666);  
    return msg_id;  
  
  
int creat_msg_queue()  
  
    int flags = IPC_CREAT | IPC_EXCL | 0666;  
    return comm_msg_queue(flags);  
  
  
int get_msg_queue()  
  
    int flags = IPC_CREAT;  
    return comm_msg_queue(flags);  
  
  
int destroy_queue(int msg_id)  
  
    if(msgctl(msg_id,IPC_RMID,NULL) != 0)  
      
        printf("%d : %s",errno,strerror(errno));  
        return -1;  
      
  
    return 0;  
  
  
int send_msg(int msg_id,int send_type,const char* msg)  
  
    struct msgbuf _buf;  
    _buf.mtype = send_type;  
    strncpy(_buf.mtext,msg,strlen(msg)+1);  
  
    if(msgsnd(msg_id,&_buf,sizeof(_buf.mtext),0) < 0)  
      
        printf("%d : %s",errno,strerror(errno));  
        return -1;  
      
    return 0;  
  
  
int recv_msg(int msg_id,int recv_type,char* msg_out)  
  
    struct msgbuf _buf;  
    _buf.mtype = 0;  
    memset(_buf.mtext,'\\0',sizeof(_buf.mtext));  
  
    if(msgrcv(msg_id,&_buf,sizeof(_buf.mtext),recv_type,0) < 0)  
      
        printf("%d : %s",errno,strerror(errno));  
        return -1;  
      
  
    strcpy(msg_out,_buf.mtext);  
    return 0;  
  

3.server.c

#include"comm.h"  
  
int main()  
  
    int msg_id = creat_msg_queue();  
    if(msg_id <0)  
      
        printf("%d : %s\\n",errno,strerror(errno));  
        return 1;  
      
  
    char buf[_SIZE_];  
    while(1)  
      
        memset(buf,'\\0',sizeof(buf));  
        recv_msg(msg_id,client_type,buf);  
        printf("client:%s\\n",buf);  
        if(strcasecmp(buf,"quit") == 0)  
          
            break;  
          
  
        printf("client say done,Please Enter# ");  
        fflush(stdout);  
        ssize_t _s = read(0,buf,sizeof(buf)-1);  
        if(_s > 0)  
          
            buf[_s - 1] = '\\0';  
          
  
        send_msg(msg_id,server_type,buf);  
  
      
  
    destroy_queue(msg_id);  
    return 0;  
  

4.client.c

#include"comm.h"  
  
int main()  
  
    int msg_id = get_msg_queue();  
  
    char buf[_SIZE_];  
    while(1)  
      
        printf("Please Enter:");  
        fflush(stdout);  
        ssize_t _s = read(0,buf,sizeof(buf)-1);  
        if(_s >0 )  
          
            buf[_s - 1] = '\\0';  
          
        send_msg(msg_id,client_type,buf);  
        if(strcasecmp(buf,"quit") == 0)  
          
            break;  
          
  
        memset(buf,'\\0',sizeof(buf));  
        recv_msg(msg_id,server_type,buf);  
        printf("server# %s\\n",buf);  
      
  
    return 0;  
  

★Makefile文件的编写：

.PHONY:all
all:client server

client:client.c comm.c       //gcc -o $@（目标文件） $^（依赖关系:后的所有内容）
gcc -o $@ $^
server:server.c comm.c
gcc -o $@ $^

.PHONY:clean
clean:
rm -f server client

★运行结果：

★注：这样就完成了client端和server端的通信。需要注意的是，运行程序时如果先打开了client端的可执行程序，此

时的第二个窗口运行server端时，可能会弹出17，FileExists的提示信息致使程序无法运行，因为此时打开client端

时已创立了一个消息队列，并申请了IPC资源。所以此时需要清除它（注意切换至root权限），并重新按照先server

后client的顺序执行即可。

命令：ipcs -q ----------------显示消息队列相关信息；

   ipcrm -q [参数]--------------这里的参数输入显示的key值或者msqid的值都可

▲温馨提示：最后make clean删除生成的可执行程序时，千万别忘了用ipcrm指令删除产生的消息队列，因为它不

具有管道那样进程退出就释放所有资源的特性，不自主手动释放，它也是会偷偷的一直存在的。

注：comm.h中的五个接口函数：

①int creat_msg_queue();  //创建一个消息队列
②int get_msg_queue();   //获得消息队列的msg_id

③int send_msg(int msg_id,int send_type,const char* msg);  //发送消息
④int recv_msg(int msg_id,int recv_type,char* msg_out);  //接收消息

⑤int destroy_queue(int msg_id);  //销毁队列