消息队列

消息队列

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

进程之间通信的方式

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

进程间通信（一）—管道传送门：http://www.cnblogs.com/lenomirei/p/5636339.html

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

更关键的是，内存他快呀，比磁盘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 be
10                                       written to queue; used to initialize
11                                       msg_qbytes during queue creation
12                                       (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 queues
17                                       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_ 0x666
14 #define _SIZE_ 1024
15 
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 msgbuf
26 {
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,'',strlen(buf)+1);
37   ssize_t _size=read(0,buf,_SIZE_);
38   if(_size>0)
39   {
40     buf[_size-1]='';
41   }
42 
43   struct msgbuf _mbuf;
44   memset(&_mbuf,'',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,'',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,'',sizeof(buf));
12     receive_msg(msg_id,c_type,buf);
13     if(strcasecmp(buf,"quit")==0)
14     {
15       break;
16     }
17     printf("client # %s
",buf);
18     printf("clent say done ! Please Input:");
19     fflush(stdout);
20     memset(buf,'',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,'',sizeof(buf));
16     send_msg(msg_id,c_type,buf);
17     if(strcasecmp(buf,"quit")==0)
18     {
19       break;
20     }
21     memset(buf,'',sizeof(buf));
22     receive_msg(msg_id,c_type,buf);
23     printf("server # %s
",buf);
24     printf("server say done ! Please Input:");
25     fflush(stdout);
26   }
27 
28   return 0;
29 }

 

分类: Operating System, C/C++