一：消息队列
消息队列是进程间通信的一种手段，进程产生的数据块以链表的形式存储在消息队列中，每个数据块都被认为是有一个类型，接收者进程接收的数据块可以有不同的类型。
消息队列有以下的几个限制，每个消息有最大长度不能超过MSGMAX，每个消息队列的总的字节数不能超过MSGMNB，消息队列的总数不能超过MSGMNI。
//查看消息队列限制的值
root@epc:~# cat /proc/sys/kernel/msgmax
8192
root@epc:~# cat /proc/sys/kernel/msgmnb
16384
root@epc:~# cat /proc/sys/kernel/msgmni
32000
root@epc:~#
二：消息队列的数据结构
常用的IPC相关linux命令：
查看ipc：ipcs
删除：ipcrm -Q //只能删除非私有ipc，私有的ipc的key为0x0
ipcrm -q
//内核为每个IPC对象维护一个数据结构
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 */
};
//linux查看IPC的命令
root@epc:~# ipcs
------ Message Queues -------- =>消息队列
key msqid owner perms used-bytes messages
------ Shared Memory Segments -------- =>共享内存
key shmid owner perms bytes nattch status
------ Semaphore Arrays -------- =>信号量
key semid owner perms nsems
//消息队列结构
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 inqueue (nonstandard) */
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) */
};
三：消息队列的函数
//头文件
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
//msgget
int msgget(key_t key, int msgfl_g);
/*
@msgget：用来创建和访问一个消息队列
@int：成功返回非负整数，是该消息队列的标识码，即id，失败返回-1
当返回-1时，errno的值有以下情况：
EACCESS(权限不允许)
EEXIST(队列已经存在，无法创建)
EIDRM(队列标志为删除)
ENOENT(队列不存在)
ENOMEM(创建队列时内存不够)
ENOSPC(超出最大队列限制)
@key：整数或者IPC_PRIVATE（0），消息队列的名字。
IPC_PRIVATE，表示私有的，意味消息队列只能用于亲缘进程通信。
@msgfl_g：九个权限标志构成，它们的用法和创建文件时使用的mode模式标志是一样的；
如果单独使用IPC_CREAT，则msgget()要么返回一个新创建的消息队列的标识符，要么返回具有相同关键字值的队列的标识符；
如果IPC_EXCL和IPC_CREAT一起使用，如果队列已经存在则返回一个失败值-1，否则创建一个新的消息队列，IPC_EXCL单独使用是没有用处的。
*/
//msgctl
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
/*
@msgget：用来查看/设置/删除消息队列
@int：成功返回0，失败返回-1
@cmd：操作类型，可取值IPC_STAT或者IPC_SET或者IPC_RMID
IPC_STAT：获取消息队列的状态，并将结果输出到buf中
IPC_SET：在进程有足够权限的条件下，把消息队列的当前关联值设置为msqid_ds数据结构中给出的值，通过导入buf的值进行设置
IPC_RMID：删除消息队列。（此时buf可置为null）
@buf：输出输入消息队列状态的空间
*/
//msgsnd
int msgsnd(int msqid, const void *msgp,size_t msgsz, int msgfl_g);
/*
@msgsnd：把一条消息添加到消息队列中去返回值
@int：成功返回0，失败返回-1
@msqid：消息队列的标志，即id
@msgp：数据块内容
数据块内容必须是如下结构：
struct msgbuf{
long mtype; //接收者函数通过这个长整数确定消息的类型
char mtext[X]; // 这里存放具体的消息内容，X必须小于等于MSGMAX
};
@msgsz：msgp中mtext长度，这个长度不含保存消息类型的那个long int长整型，此值必须小于MSGMAX，否则返回-1
@msgfl_g：控制位，控制当前消息队列满或到达系统上限时将要发生的事情。
为0表示等待；msgfl_g =IPC_NOWAIT。表示队列满不等待，返回EAGAIN。
*/
//msgrcv
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgfl_g)
/*
@msgrcv：从一个消息队列中接收消息
@ssize_t ：成功返回实际接收到放到接收缓冲区里去的字符个数，失败返回-1
@msqid：消息队列的标志，即id
@msgp：数据块内容输出
@msgsz：msgp指向的消息长度，不包含消息类型的长度
@msgtyp：接收的消息类型
=0：表示返回队列里的第一条消息
>0：返回队列第一条类型等于msgtype的消息
<0：返回队列第一条类型小于等于msgtype绝对值的消息
@msgfl_g：控制位，控制着队列中没有相应类型的消息可工接收时将要发生的事情。
msgfl_g=IPC_NOWAIT：队列没有可读消息不等待，返回ENOMSG错误
msgfl_g=MSG_NOERROR：消息大小超过msgsz时被截断
msgtype>0且msgfl_g=MSG_EXCEPT：接收类型不等于msgtype的第一条消息
*/
四：利用System V消息队列实现简单的回射客户端/服务器
//server
#include<stdio.h>
#include<sys/types.h>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<stdlib.h>
#include<string.h>
#define ERR_EXIT(m) \
do \
{ \
perror(m); \
exit(EXIT_FAILURE); \
}while(0)
#define MSGMAX 8192
struct Msgbuf{
long mtype;
char mtext[MSGMAX];
};
void echo_ser(int msgid)
{
struct Msgbuf msg;
memset(&msg, 0, sizeof(msg));
int nrcv = 0;
while (1)
{
if ((nrcv = msgrcv(msgid, &msg, MSGMAX, 1, 0)) < 0)
ERR_EXIT("msgrcv");
int pid = *((int *)msg.mtext);
printf("recv from pid=%d, length=%d, buf=%s\n", pid, nrcv, msg.mtext+4);
msg.mtype = pid;
msgsnd(msgid, &msg, nrcv, 0);
memset(&msg, 0, sizeof(msg));
}
}
int main(int argc, char *argv[])
{
int msgid;
msgid = msgget(1234, IPC_CREAT | 0666);
if (msgid == -1)
ERR_EXIT("msgget");
echo_ser(msgid);
return 0;
}
//client
#include<stdio.h>
#include<stdlib.h>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<sys/types.h>
#include<unistd.h>
#include<errno.h>
#include<string.h>
#define ERR_EXIT(m) \
do { \
perror(m); \
exit(EXIT_FAILURE); \
} while(0)
#define MSGMAX 8192
struct Msgbuf
{
long mtype;
char mtext[MSGMAX];
};
void echo_cli(int msgid)
{
int nrcv;
int pid = getpid();
struct Msgbuf msg;
memset(&msg, 0, sizeof(msg));
while (fgets(msg.mtext+4, MSGMAX, stdin) != NULL)
{
msg.mtype = 1;
*((int *)msg.mtext) = pid;
if (msgsnd(msgid, &msg, 4 + strlen(msg.mtext + 4), IPC_NOWAIT) < 0)
ERR_EXIT("msgsnd");
if ((nrcv = msgrcv(msgid, &msg, MSGMAX, pid, 0)) < 0)
ERR_EXIT("msgsnd");
fputs(msg.mtext + 4, stdout);
memset(&msg, 0, sizeof(msg));
}
}
int main(int argc, char *argv[])
{
int msgid;
msgid = msgget(1234, 0);
if (msgid == -1)
ERR_EXIT("msgget");
echo_cli(msgid);
return 0;
}
五：实现每个客户一个队列
//server
#include<stdio.h>
#include<sys/types.h>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<stdlib.h>
#include<string.h>
#include<signal.h>
#include<wait.h>
#include<unistd.h>
#include<errno.h>
#define ERR_EXIT(m) \
do \
{ \
perror(m); \
exit(EXIT_FAILURE); \
}while(0)
#define MSGMAX 8192
#define MSGQ_KEY 1234
#define MTYPE_GLOBAL 1
struct Msgbuf{
long mtype;
char mtext[MSGMAX];
};
void Handler(int signo)
{
pid_t pid;
int stat;
while ((pid = waitpid(-1, &stat, WNOHANG)) > 0);
return;
}
/*
* void echo_ser(int msgid);
* @echo_ser：实现用单独的队列和每个客户进行交互，为了避免单线程处理一个客户信息等待时间过长，引入了用子进程进行处理的机制；
*
* */
void echo_ser(int msgid)
{
signal(SIGCHLD, Handler);
struct Msgbuf msg;
memset(&msg, 0, sizeof(msg));
int nrcv = 0;
while (1)
{
do
{
//处理子进程退出时，会发生中断，导致msgrcvc返回错误；
nrcv = msgrcv(msgid, &msg, MSGMAX, MTYPE_GLOBAL, 0);
}while(nrcv<0 && errno==EINTR);
if (nrcv < 0)
ERR_EXIT("msgrcv");
int clientMsgid = *((int *)msg.mtext);
printf("recv from clientMsgid=%d, length=%d, buf=%s", clientMsgid, nrcv, msg.mtext+4);
pid_t pid = fork();
if (pid < 0)
ERR_EXIT("fork");
else if (pid == 0)
{
msg.mtype = MTYPE_GLOBAL;
msgsnd(clientMsgid, &msg, nrcv, IPC_NOWAIT);
memset(&msg, 0, sizeof(msg));
break;
}
memset(&msg, 0, sizeof(msg));
}
}
int main(int argc, char *argv[])
{
int msgid;
msgid = msgget(MSGQ_KEY, IPC_CREAT | 0666);
if (msgid == -1)
ERR_EXIT("msgget");
echo_ser(msgid);
return 0;
}
//客户端
#include<stdio.h>
#include<stdlib.h>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<sys/types.h>
#include<unistd.h>
#include<errno.h>
#include<string.h>
#define ERR_EXIT(m) \
do { \
perror(m); \
exit(EXIT_FAILURE); \
} while(0)
#define MSGMAX 8192
#define MSGQ_KEY 1234
#define MTYPE_GLOBAL 1
struct Msgbuf
{
long mtype;
char mtext[MSGMAX];
};
void echo_cli(int msgidForWr, int msgidForRead)
{
int nrcv;
int pid = getpid();
struct Msgbuf msg;
memset(&msg, 0, sizeof(msg));
while (fgets(msg.mtext+4, MSGMAX, stdin) != NULL)
{
msg.mtype = MTYPE_GLOBAL;
*((int *)msg.mtext) = msgidForRead;
if (msgsnd(msgidForWr, &msg, 4 + strlen(msg.mtext + 4), IPC_NOWAIT) < 0)
ERR_EXIT("msgsnd");
if ((nrcv = msgrcv(msgidForRead, &msg, MSGMAX, MTYPE_GLOBAL, 0)) < 0)
ERR_EXIT("msgsnd");
fputs(msg.mtext + 4, stdout);
if (msg.mtext[4] == '\n') break;
memset(&msg, 0, sizeof(msg));
}
msgctl(msgidForRead, IPC_RMID, NULL);
}
int main(int argc, char *argv[])
{
int msgidForWr, msgidForRead;
msgidForWr = msgget(MSGQ_KEY, IPC_CREAT);
if (msgidForWr == -1)
ERR_EXIT("msgget");
msgidForRead = msgget(IPC_PRIVATE, 0666 | IPC_CREAT |IPC_EXCL);
if (msgidForWr == -1)
ERR_EXIT("msgget");
echo_cli(msgidForWr, msgidForRead);
return 0;
}