消息队列

上次说到了进程间通信的管道，不过匿名管道有个缺点就是，只能做到有亲缘关系的进程间通信，所以今天学习一个新的进程间通信方式——消息队列。

• 消息队列提供了一个从一个进程向另外一个进程发送一块数据的方法
• 每个数据块都被认为是有一个类型，接收者进程接收的数据块可以有不同的类型值

• 消息队列也有管道一样的不足，就是每个数据块的最大长度是有上限的，系统上全体队列的最大总长度也有一个上限
  

  消息队列函数

  头文件
  #include < sys/types.h>
  #include < sys/ipc.h>
  #include < sys/msg.h>

  msgget

  int msgget(key_t key, int msgflg);
  作用：创建和访问一个消息队列
  key：某个消息队列的名字（类似于每个进程都有一个进程ID一样）
  msgflg：有9个权限标志构成。它们的用法和创建文件时使用的mode标志是一样的（比如：一个key已经存在的消息队列时，要使用IPC_CREAT | IPC_EXCL，就类似于文件操作的打开：O_CREAT | O_EXCL ）。
  返回值：成功将返回一个非负整数，即该消息队列的标识码；失败返回-1

  msgsnd

  int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

  作用：把一条消息加到消息队列里面
  msgid：由msgget函数返回的消息队列标识码
  msgp：是一个指针，指向准备发送的消息
  msgsz：msgp指向的消息长度，这个长度不能保存消息类型里的“long int”类型（下面会说）
  msgflg：控制着当前消息队列满或达到系统上限时将要发生的事情。
  返回值：成功-0，失败-1
  1）

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

  消息的指针就是指向这样一个结构的消息，这需要我们自己定义。但是，第一个一定得是long int，表示消息的类型。消息的类型是大于0的整数（当然也是可以等于0的，但是这样就意味着任何消息我都接收，不固定只收某一种类型的消息了）。

• msgtype=0返回队列第一条信息
• msgtype>0返回队列第一条类型等于msgtype的消息
• msgtype< 0返回队列第一条类型小于等于msgtype绝对值的消息

msgrcv

作用：从一个消息队列里检索消息
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
msgid：由msgget函数返回的消息队列标识码
msgp：是一个指针，指向准备接收的消息
msgsz：msgp指向的消息长度，这个长度不能保存消息类型里的“long int”类型（下面会说）
msgflg：控制着队列中没有相应类型的消息可供接收的时候将要发生的事
msgtyp：可以实现接收优先级的简单形式
返回值：成功-返回实际放到接收缓冲区里的字符个数，失败- “-1”
msgflg有以下几个值：

• msgflg=IPC_NOWAIT，队列没有可读消息不等待，返回ENOMSG错误。
• msgflg=MSG_NOERROR，消息大小超过msgsz时被截断
• msgtype>0且msgflg=MSC_EXCEPT，接收类型不等于msgtype的第一条消息。

  msgctl

  int msgctl(int msqid, int cmd, struct msqid_ds *buf);

  作用：消息队列的控制函数
  msgid：由msgget函数返回的消息队列标识码
  cmd：将要采取的动作，简单讲常用的三个可取值：

msgid_ds数据结构定义如下：

 struct msqid_ds { struct ipc_perm msg_perm; /* Ownership and permissions */ time_t msg_stime; /* Time of last msgsnd(2) */ time_t msg_rtime; /* Time of last msgrcv(2) */ time_t msg_ctime; /* Time of last change */ unsigned long __msg_cbytes; /* Current number of bytes in queue (non-standard) */ msgqnum_t msg_qnum; /* Current number of messages in queue */ msglen_t msg_qbytes; /* Maximum number of bytes allowed in queue */ pid_t msg_lspid; /* PID of last msgsnd(2) */ pid_t msg_lrpid; /* PID of last msgrcv(2) */ };

练习

试着写一段吧~
发送端：msgsnd.c

struct msg_t { long mtype; //第一个必须是long,>=1 char acMsg[20]; }; //用ipcs int main() { int msgid; struct msg_t msg = {0}; //消息队列的创建、打开、删除 msgid = msgget(1000,IPC_CREAT); if(msgid == -1) { perror("create msg"); } printf("msgid = %d\n",msgid); msg.mtype = 1; strcpy(msg.acMsg,"hello"); msgsnd(msgid,&msg,sizeof(struct msg_t)-sizeof(long),0); //msgctl(msgid,IPC_RMID,NULL);//也可以用命令ipcrm return 0; }

接收端：msgrcv.c

struct msg_t { long mtype; //第一个必须是long,>=1 char acMsg[20]; }; int main() { int msgid; struct msg_t msg = {0}; //消息队列的创建、打开、删除 msgid = msgget(1000,0); if(msgid == -1) { perror("open msg!\n"); } printf("msgid = %d\n",msgid); msgrcv(msgid,&msg,sizeof(struct msg_t)-sizeof(long),1,0); printf("recv msg: %s.\n",msg.acMsg); return 0; }

运行：
开三个终端，一个运行msgsnd.c，一个运行msgrcv.c，一个用来查看消息队列状态：
1、先发送：

2、查看：ipcs

3、读取：

4、再查看（被取走了）：

<br>
但是，也注意到了使用消息队列，消息一旦被读走，就没了。

即时通讯小程序

使用消息队列与共享内存（后面会复习）完成一个简单的终端聊天程序，要求如下。
1.程序有一个server服务器，服务器有一个在线列表，当终端登入时，将终端的进程ID作为用户名称添加到在线列表。（消息队列和共享内存）
2.终端登入时，获取用户的在线列表。（消息队列、信号、和共享内存）
3.终端登入后，进入聊天状态。：（信号、共享内存）
输入#chat [pid],进入私聊模式
例如：#chat 1234，与终端1234进入私聊。只有终端1234才能接收消息。
输入#chat 0,进入群聊模式。全部终端可以接收消息。
说明：1,2,3为基本功能，要求实现。4,5,6为附加功能，有能力同学可以尝试实现。
4.终端登入时，服务器发送消息，通知其他在线终端，更新在线列表。（信号）
5.终端在聊天状态，输入#user，列出在线用户列表。
6.终端在聊天状态，输入#exit，终端退出，服务器将终端的进程ID移出在线列表，并通知在线终端，更新在线列表。(消息队列和共享内存)

学了消息队列，至少可以把用户的登录、退出完成。
代码如下：
public.h

#ifndef _PUBLIC_H_ #define _PUBLIC_H_ #include < stdio.h> #include < string.h> #include < sys/types.h> #include < sys/ipc.h> #include < sys/msg.h> typedef struct login_t { long type; pid_t pid; }LOGIN_T; #define MSG_KEY 1 #define MSG_SIZE sizeof(LOGIN_T)-sizeof(long) #endif

server.c

#include " public.h" int main() { int msg_id; LOGIN_T login = {0}; //创建用户的消息队列 msg_id = msgget(MSG_KEY,0); if(msg_id == -1) { msg_id = msgget(MSG_KEY,IPC_CREAT); if (msg_id == -1) { perror("server msgget"); return -1; } } //一直监听，是否有用户上线 while (1) { memset(&login,0,sizeof(LOGIN_T)); msgrcv(msg_id,&login,MSG_SIZE,0,0); //任何消息都接收 switch(login.type) { case 1: printf("client %d is logining...\n",login.pid); break; case 2: printf("client %d is exiting...\n",login.pid); break; } } return 0; }

client.c

#include "public.h" int main() { char acBuf[20] = ""; int msg_id; LOGIN_T login = {0}; //打开消息队列 msg_id = msgget(MSG_KEY,0); if(msg_id == -1) { perror("client msgget"); return -1; } //登录，写消息队列 login.type = 1; //设置登录的消息类型为1 login.pid = getpid(); printf("%d is logining...\n",login.pid); msgsnd(msg_id,&login,MSG_SIZE,0); //等待写 while(1) { putchar('#'); fflush(stdout); scanf("%s",acBuf); if (strcmp(acBuf,"quit") == 0) { login.type = 2; //设置退出的消息类型为2 msgsnd(msg_id,&login,MSG_SIZE,0); break; } } return 0; }

运行：先运行服务器端，再运行多个客户端
客户端：

服务器：