你的位置:首页 > 操作系统

[操作系统]进程间通信(二)—消息队列


我会用几篇博客总结一下在Linux中进程之间通信的几种方法,我会把这个开头的摘要部分在这个系列的每篇博客中都打出来

进程之间通信的方式

  • 管道
  • 消息队列
  • 信号
  • 信号量
  • 共享存储区
  • 套接字(socket)

这次主要写的是消息队列,之前讲过的管道和消息队列在本质上就有很大的区别,管道是一个文件,而消息队列是一个数据结构(类似于链表)。这说明了,管道文件是存放在磁盘上的,关机也会存在(尤其是命名管道更为显而易见,你不删除他他就搁那呆着),而消息队列是存在于内核中的内存,显而易见,关机就没了。

更关键的是,内存他快呀,比磁盘I/O快多了,为啥要用那么慢的管道。而且消息队列是可以直接完成没有亲缘关系的进程之间的通信的。但是结构比起管道要复杂,用到了很多结构体。内容有些多。先写一下和管道的主要区别,可以更直观的进行对比

  • 匿名管道是跟随进程的,消息队列是跟随内核的,也就是说进程结束之后,匿名管道就死了,但是消息队列还会存在(除非显示调用函数销毁)
  • 管道是文件,存放在磁盘上,访问速度慢,消息队列是数据结构,存放在内存,访问速度快
  • 管道是数据流式存取,消息队列是数据块式存取

那么从头开始吧:

 

  • 如何创建一个消息队列

在C库函数中有一个系统调用可以创建一个消息队列,那就是msgget,跟这个函数有关的其他函数也会一并给出来

  • 函数原型: int msgget(key_t key, int msgflg)
  • 头文件:#include <sys/types.h> #include <sys/ipc.h>  #include <sys/msg.h>
  • 参数解析
    • 第一个参数是一个标识数据结构唯一的key键值,可以给IPC_PRIVATE让内核自动给,也可以自己调用ftok函数绑定一个
    • 第二个参数是创建消息队列的参数,有IPC_CREAT 和 IPC_EXCL
      • 单独使用IPC_CREAT,如果该消息队列已经存在(就是该key_t对象已经拿去被创建过一个队列了),打开该队列并返回,如果不存在,就创建一个返回
      • 单独使用IPC_EXCL没有意义
      • 两个参数一起使用(IPC_CREAT | IPC_EXCL),如果该队列存在,出错返回,如果不存在创建一个返回,也就是说这样使用一定会获得一个新队列
  • 返回值,成功返回标志消息队列的唯一的一个int,失败返回-1

 

  • 函数原型:key_t ftok(const char *pathname,int proj_id)
  • 头文件:#include <sys/types.h>  #include <sys/ipc.h>
  • 参数解析
    • 没啥好解析的,第一个参数就是给个路径(目录)就成了
    • 第二个就是给个int值,没啥特殊要求,ftok本质就是把这个proj_id和pathname绑定在一起罢了

相关函数就这么多,到这里就可以开心的创建一个新的消息队列了,那么怎么看我们创建出来的消息队列呢?怎么销毁他呢?

先用命令看一下,使用ipcs -q就可查看消息队列的状态,这里我创建了一个消息队列,下面来销毁它

使用ipcrm -q msqid就可以销毁一个消息队列,我已经把刚才创建的消息队列销毁了

那么如何通过C函数销毁一个消息队列?使用msgctl函数

  • 函数原型: int msgctl(int msgid ,int cmd ,struct msgid_ds *buf)
  • 头文件:#include <sys/types.h>  #include <sys/ipc.h>  #include <sys/msg.h>
  • 参数解析
    • 第一个参数就是创建好的标识msg的int变量
    • 第二个参数有很多,这里介绍两个,先说主要的销毁,IPC_RMID,设置了这个之后第三个参数就没必要设置了给个0就成
    • 第二个参数设置成IPC_SET的时候表示对消息队列进行初始化,这时第三个参数就有用了,但是我这里没用到就不话费篇幅详细解释了,贴出结构看看就成

这里提一下,消息队列的一个缺点,这里说到IPC_SET对消息队列的初始化,就是说消息队列的创建和初始化是分开的,这样设计不是很好,因为有线程安全的问题,当一个线程创建了消息队列还没初始化,另一个线程直接开始执行访问的操作就很尴尬了

 1 struct msginfo { 2            int msgpool; /* Size in kibibytes of buffer pool 3                    used to hold message data; 4                    unused within kernel */ 5            int msgmap; /* Maximum number of entries in message 6                    map; unused within kernel */ 7            int msgmax; /* Maximum number of bytes that can be 8                    written in a single message */ 9            int msgmnb; /* Maximum number of bytes that can be10                    written to queue; used to initialize11                    msg_qbytes during queue creation12                    (msgget(2)) */13            int msgmni; /* Maximum number of message queues */14            int msgssz; /* Message segment size;15                    unused within kernel */16            int msgtql; /* Maximum number of messages on all queues17                    in system; unused within kernel */18            unsigned short int msgseg;19                  /* Maximum number of segments;20                    unused within kernel */21          };

 

  • 使用消息队列

提到使用消息队列无非就是消息的写入和消息的读出,这涉及两个函数和一个结构体

先说发送函数

  • 函数原型:int msgsnd(int msgid,const void *msgp,size_t msgsz,int msgflg)
  • 头文件:#include <sys/types.h>  #include <sys/ipc.h>   #include <sys/msg.h>
  • 参数解析
    • 第一个还是消息队列号
    • 第二个是一个结构体的指针,这个结构体叫做msgbuf在内核中有(应该是有的。。。没有就自己写一个放程序里),必要的,这个就是传输数据的数据块,没了它就没得传了,可不能随便设置为0
    • 第三个就是传输的消息长度(一般你的数据有多长就写多长,不是msgbuf对象的总大小)
    • 第四个是传送方式(无阻塞啊什么的)的参数,我的程序没有使用所以我设置为0了

在列接受函数之前必须把结构体放出来,不然下面就懵逼了

1 struct msgbuf {2        long mtype;    /* message type, must be > 0 */3        char mtext[1];  /* message data */4      };

mtype讲起,因为队列是好多进程(或者说是线程)都拿来用的,比如4个进程共用一个消息队列,两两之间进行通信,就有必要标识哪两个进程是一组的,哪块数据是属于你们组的,不能乱拿是不是,这个mtype就是用来标识的,但是一定要大于0,我一开始设置为0一直报参数错误,查了文档才发现自己蠢了。。。

mtext存放的就是你要传输的数据了,怎么大小只有1?当然不能只有1,我自己写的msgbuf就有1024长度。。。所以还是自己写爽啊

 可以说接受函数了

  • 函数原型:ssize_t msgrcv(int msgid,const void *msgp,size_t msgsz,long msgtyp,int msgflg)
  • 头文件同msgsnd函数
  • 参数解析
    • 只用看标红部分就成了,其他的和msgsnd一样
    • msgtyp就是用来表示我(当前进程)拿数据的时候,只拿(msgbuf对象中的mtype)和我传入的msgtyp一样的数据块(msgbuf对象)
  •  成功返回长度,失败返回-1

 

事已至此,基本操作就说完了,废话少说,show me the code

我的程序分为comm.h(公共头文件)  comm.c(封装基本函数) server.c(建议服务器端) client.c(简易客户端)一共4个文件

完成了服务器端和客户端的简单通信(回合制聊天(误))

comm.h

 1 #include <stdio.h> 2 #include <unistd.h> 3 #include <sys/msg.h> 4 #include <sys/types.h> 5 #include <sys/ipc.h> 6 #include <string.h> 7 #include <stdlib.h> 8 #include <errno.h> 9 #include <memory.h>10 11 12 #define _PATH_NAME_ "/tmp"13 #define _PROJ_ID_ 0x66614 #define _SIZE_ 102415 16 static int comm_create_msg_set(int flags);17 int create_msg_set();18 int get_msg_set();19 void destory_msg_set(int msg_id);20 void send_msg(int msg_id,long msgtype,char * buf);21 void receive_msg(int msg_id,long msgtype,char *buf);22 23 24 25 struct msgbuf26 {27  long mtype;28  char mtext[_SIZE_];29 };

comm.c

 1 #include "comm.h" 2 static int comm_create_msg_set(int flags) 3 { 4  key_t _key=ftok(_PATH_NAME_,_PROJ_ID_); 5  if(_key<0) 6  { 7   printf("%d:%s",errno,strerror(errno)); 8  } 9  int msg_id=msgget(_key,flags);10  if(msg_id<0)11  {12   printf("%d:%s",errno,strerror(errno));13  }14  return msg_id;15 }16 17 18 int get_msg_set()19 {20  key_t _key=ftok(_PATH_NAME_,_PROJ_ID_);21  int flags=IPC_CREAT;22  return comm_create_msg_set(flags);23 }24 25 26 int create_msg_set()27 {28  key_t _key=ftok(_PATH_NAME_,_PROJ_ID_);29  int flags=IPC_CREAT | IPC_EXCL;30  return comm_create_msg_set(flags);31 }32 33 34 void send_msg(int msg_id,long msgtype,char *buf)35 {36  memset(buf,'\0',strlen(buf)+1);37  ssize_t _size=read(0,buf,_SIZE_);38  if(_size>0)39  {40   buf[_size-1]='\0';41  }42 43  struct msgbuf _mbuf;44  memset(&_mbuf,'\0',sizeof(struct msgbuf));45  _mbuf.mtype=msgtype;46  strcpy(_mbuf.mtext,buf);47  if(msgsnd(msg_id,&_mbuf,_size,0)<0)48  {49   printf("send error,%d:%s",errno,strerror(errno));50  }51 }52 53 void receive_msg(int msg_id , long msgtype ,char *buf)54 {55  struct msgbuf _mbuf;56  memset(&_mbuf,'\0',sizeof(struct msgbuf));57  _mbuf.mtype=0;58  if(msgrcv(msg_id,&_mbuf,_SIZE_,msgtype,0)<0)59  {60   printf("recv error %d:%s",errno,strerror(errno));61  }62  strcpy(buf,_mbuf.mtext);63 }64 65 void destory_msg_set(int msg_id)66 {67  if(msgctl(msg_id,IPC_RMID,0)<0)68  {69   printf("%d:%s",errno,strerror(errno));70  }71 }

server.c

 1 #include "comm.h" 2 long c_type=1; 3 long s_type=22; 4 int main() 5 { 6  int msg_id=create_msg_set(); 7  char buf[_SIZE_]; 8  9  while(1)10  {11   memset(buf,'\0',sizeof(buf));12   receive_msg(msg_id,c_type,buf);13   if(strcasecmp(buf,"quit")==0)14   {15    break;16   }17   printf("client # %s\n",buf);18   printf("clent say done ! Please Input:");19   fflush(stdout);20   memset(buf,'\0',sizeof(buf));21   send_msg(msg_id,c_type,buf);22  }23  destory_msg_set(msg_id);24  return 0;25 }

client.c

 1 #include "comm.h" 2  3  4 long c_type=1; 5 long s_type=2; 6  7 int main() 8 { 9  int msg_id = get_msg_set();10 11  char buf[_SIZE_];12 13  while(1)14  {15   memset(buf,'\0',sizeof(buf));16   send_msg(msg_id,c_type,buf);17   if(strcasecmp(buf,"quit")==0)18   {19    break;20   }21   memset(buf,'\0',sizeof(buf));22   receive_msg(msg_id,c_type,buf);23   printf("server # %s\n",buf);24   printf("server say done ! Please Input:");25   fflush(stdout);26  }27 28  return 0;29 }