消息队列 ---- 双向通信（读取不一定先入先出）
1、 消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。 每个数据块都被认
为是有一个类型，接收者进程接收的数据块可以有不同的类型值。我们可以通过发送消息
来避免命名管道的同步和阻塞问题。
消息队列与管道不同的是，消息队列是基于消息的，而管道是基于字节流的，且消息队列的读取不一定是先入先出。
2、消息队列的结构：
struct ipc_perm
{
key_t _key; /* Key supplied to xxxget(2) */
uid_t uid; /* Effective UID of owner */
gid_t gid; /* Effective GID of owner */
uid_t cuid; /* Effective UID of creator */
gid_t cgid; /* Effective GID of creator */
unsigned short mode; /* Permissions */
unsigned short _seq; /* Sequence number */
};
3、消息队列的实现：
头文件：
#include<sys/types.h>
#include<sys/ipc.h>
#include<sys/msg.h>
--------------------------------------------------------------------
（1）创建 新 消息队列 或 取得 已存在的消息队列
int msgget(key_t key, int msgflg);//创建新消息队列或取得已存在消息队列
key：可以认为是一个端口号，也可以由函数ftok()生成。
msgflg：
IPC_CREAT 如果IPC不存在，则创建一个IPC资源，否则打开操作。
IPC_EXCL：只有在共享内存不存在的时候，新的共享内存才建立，否则就产生错误。
（2）向 队列 读/写 消息
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);//从队列中读消息 int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);//将数据写入消息队列中
msgflg：指明核心程序在队列没有数据的情况下所应采取的行动。
如果msgflg和常数IPC_NOWAIT合用，则：1.在msgsnd()执行时若是消息队列已满，则msgsnd()将不会阻塞，而会立即返回-1；2.如果执行的是msgrcv()，则在消息队列呈空时，不做等待马上返回-1，并设定错误码为ENOMSG。
当msgflg为0时，msgsnd()及msgrcv()在队列呈满或呈空的情形时，采取
阻塞等待的处理模式。
（3）设置 消息队列 属性
int msgctl ( int msgqid, int cmd, struct msqid_ds *buf );//设置消息队列属性
cmd 操作： IPC_STAT , IPC_SET , IPC_RMID
IPC_STAT : 获取消息队列对应的 msqid_ds 数据结构，并将其保存到 buf 指定的地址空间。
IPC_SET : 设置消息队列的属性，要设置的属性存储在buf中。
IPC_RMID : 从内核中删除 msqid 标识的消息队列。
4、实例：client server 双向通信
//comm.h
#ifndef _MSG_
#define _MSG_
#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<string.h>
#include<sys/types.h>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<errno.h>
#define _PATH_ "."
#define _PROC_ID_ 0x6666
#define _CLIENT_ID_ 1
#define _SERVER_ID_ 2
#define _BLOCK_SIZE_ 256
struct msg_buf
{
long mtype;
char mtext[_BLOCK_SIZE_];
};
int comm_msg_queue(int flag);
int create_msg_queue();
int get_msg_queue();
int msg_send(int msg_id,int _mtype,const char *_info);
int msg_recv(int msg_id,int recv_type,char buf[]);
int destroy_msg_queue(int msg_id);
#endif
//comm.c
#include"comm.h"
int comm_msg_queue(int flag)
{
key_t _key=ftok(_PATH_,_PROC_ID_);
if(_key < 0)
{
perror("ftok");
return -1;
}
int msg_id=msgget(_key,flag);
if(msg_id < 0)
{
perror("msgget");
return -1;
}
return msg_id;
}
int create_msg_queue()
{
int flag = (IPC_CREAT |IPC_EXCL |0666); //0666 ：set permission
return comm_msg_queue(flag);
}
int get_msg_queue()
{
int flag = IPC_CREAT;
return comm_msg_queue(flag);
}
int msg_send(int msg_id,int _mtype,const char *_info)
{
struct msg_buf _msg;
_msg.mtype=_mtype;
memset((void*)_msg.mtype,‘\0‘,sizeof(_msg.mtype));
strcpy(_msg.mtext,_info);
if(msgsnd(msg_id,&_msg,sizeof(_msg.mtext),IPC_NOWAIT) < 0)//non-blocking wait
{
perror("msgsnd");
return -1;
}
return 0;
}
int msg_recv(int msg_id,int recv_type,char buf[])
{
struct msg_buf _msg;
if(msgrcv(msg_id,&_msg,sizeof(_msg.mtext),recv_type,0) < 0)//blocking wait
{
perror("msgrcv");
return -1;
}
strcpy(buf,_msg.mtext);
return 0;
}
int destroy_msg_queue(int msg_id)
{
if(msgctl(msg_id,IPC_RMID,NULL) < 0)//release
{
perror("msgctl");
return -1;
}
else
{
printf("remove msg queue done ...\n");
}
return 0;
}
//server.c
#include"comm.h"
int main()
{
int queue_id=create_msg_queue();
if(queue_id < 0)
{
perror("create_msg_queue");
exit(1);
}
char buf[_BLOCK_SIZE_];
while(1)
{
memset(buf,‘\0‘,sizeof(buf));
msg_recv(queue_id,_CLIENT_ID_,buf);//server receives info from client
printf("Client #%s\n",buf);
printf("Please Enter:");
fflush(stdout);
fgets(buf,sizeof(buf),stdin);
msg_send(queue_id,_SERVER_ID_,buf);//server sends info to client
}
destroy_msg_queue(queue_id);
return 0;
}
//client.c
#include"comm.h"
int main()
{
int queue_id=get_msg_queue();
if(queue_id < 0)
{
perror("get_msg_queue");
exit(1);
}
char buf[_BLOCK_SIZE_];
while(1)
{
printf("Please Enter:");//input info
fflush(stdout);
fgets(buf,sizeof(buf),stdin);
msg_send(queue_id,_CLIENT_ID_,buf);//client sends info to server
msg_recv(queue_id,_SERVER_ID_,buf);//client receives info from server
printf("Server #%s\n",buf);
}
destroy_msg_queue(queue_id);
return 0;
}