定义
消息队列本质是消息的链接表，存储在内核中，由消息队列标识符标识。消息队列有两种版本，一种是system V4版本，一种是POSIX IPC版本。消息队列顾名思义，当一个进程需要向另一个进程传递消息时，则将需要传递的数据存放在特定的队列中，另一个进程则从该队列中取出即可。原理与管道类视，只是两者的实现方式不同，函数接口不同。
队列ID是消息队列的核心变量，消息队列通过队列ID来识别，该将消息存放在哪个队列，从哪个队列中读取消息。
数据结构
- system V版本相关的数据结构
struct msqid_ds {
struct ipc_perm msg_perm; /* Ownership and permissions */
time_t msg_stime; /* Time of last msgend() */
time_t msg_rtime; /* Time of last msgrcv() */
unsigned long __msg_cbytes; /* Current number of bytes in
queue (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() */
pid_t msg_lrpid; /* PID of last msgrcv() */
}
struct ipc_perm {
key_t __key; /* Key supplied to msgget() */
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 */
};
- POSIX版本相关的数据结构
struct mq_attr {
long mq_flags; /* Flags (ignored for mq_open()) */
long mq_maxmsg; /* Max. # of messages on queue */
long mq_msgsize; /* Max. message size (bytes) */
long mq_curmsgs; /* # of messages currently in queue (ignored for mq_open()) */
};
system V 函数接口
- 创建消息队列
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);
key是一个长整型数，这个键值有内核转换为标识符
msgflg，标志用于指示队列的权限位，类似于创建fd时的flag
返回值，返回队列ID
- 控制消息队列
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
msqid，队列ID，指定需要设置的队列
cmd，对队列执行的指令
IPC_STAT:获取消息队列的属性信息
IPC_SET:设置消息队列的属性
IPC_RMID:删除消息队列
msqid_ds存储队列相关属性的结构体
- 发送消息队列
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg)
msqid，队列ID，指定发送消息的目的队列
msgp,指向要发送的数据的buffer指针，该指针指向的数据需要具备一定的格式，如下msgbuf
struct msgbuf{
long mtype; /* 消息类型，必须大于0 */
char mtext[1]; /* 消息数据，由编程者根据场景决定数组大小，最大可为512字节 */
}
msgsz，发送消息的数据大小
msgflag，发送消息的标志位，比如是阻塞发送还是不阻塞发送，当队列满的时候，不阻塞会马上返回错误，阻塞发送会一直等待队列空闲再发送
- 接收消息队列
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);
msqid，队列ID，指定接收消息的队列
msgp，指定接收的buffer
msgsz，指定接收数据大小
msgtype，指定接收消息类型
msgflg，指定标志位
system V消息队列的局限性：
当队列来消息时，不能主动通知进程，需要接收消息的进程只能阻塞等待消息来后再接收，或者选择不阻塞接收，当没有消息时候直接返回，靠周期轮询接收消息，这样导致接收消息效率低。
ipcs命令可以获取系统中IPC的相关信息
POSIX 函数接口
POSIX版本遵循一切文件的理念，按照文件的形式实现消息队列，所呈现的函数接口都是类似文件I/O的接口，如open、close、unlink等。
相比system V版本POSIX版本消息队列还提供了通知的功能，当消息队列来消息时，就会通过信号通知对应的进程。
- 打开或创建消息队列
#include <fcntl.h>
#include <sys/stat.h>
#include <mqueue.h>
mqd_t mq_open(const char *name, int oflag, mode_t mode,struct mq_attr *attr);
Link with -lrt.
name，队列名字，需要包含一个'/'字符，不然会出现EINVAL invalid argument 非法参数的报错
oflag，控制发送接收属性，O_RDONLY，O_WRONLY，O_RDWR，O_CREAT，O_NONBLOCK等
mode，一个可选参数，只有在创建消息队列时才需要，表示默认的访问权限
attr，一个可选参数，创建新消息队列时才需要，设置新消息队列的属性
返回队列描述符
消息队列创建成功后，系统会在/dev/mqueue目录下生成对应的文件，里面保存着消息队列的相关信息
- 关闭消息队列
#include <mqueue.h>
int mq_close(mqd_t mqdes);
- 删除消息队列
#include <mqueue.h>
int mq_unlink(const char *name);
- 获取消息队列属性
#include <mqueue.h>
int mq_getattr(mqd_t mqdes,struct mq_attr *attr);
mqdes，队列描述符
attr，队列属性
- 修改消息队列属性
#include <mqueue.h>
int mq_setattr(mqd_t mqdes,const struct mq_attr *newattr,struct mq_attr *oldattr);
mqdes，队列描述符
newattr，队列新属性
oldattr，队列旧属性
- 发送消息队列
#include <mqueue.h>
#include <time.h>
int mq_send(mqd_t mqdes, const char *msg_ptr,
size_t msg_len, unsigned int msg_prio);
int mq_timedsend(mqd_t mqdes, const char *msg_ptr,size_t msg_len, unsigned int msg_prio,const struct timespec *abs_timeout)
mqdes，队列描述符
msg_ptr，指向消息队列数据的指针
msg_len，发送数据的大小
msg_prio，消息的优先级
abs_timeout，阻塞等待的时间
- 接收消息队列
#include <time.h>
#include <mqueue.h>
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr,size_t msg_len, unsigned int *msg_prio);
ssize_t mq_timedreceive(mqd_t mqdes, char *msg_ptr,size_t msg_len, unsigned int *msg_prio,const struct timespec *abs_timeout);
mqdes，队列描述符
msg_ptr，指向消息队列数据的指针
msg_len，接收数据的大小，这里应该传入最大的队列数据长度，不然会出现too long messege
msg_prio，消息的优先级
abs_timeout，阻塞等待的时间
因为mq_XXX()系列函数不是标准C库函数，链接时需要指定库-lrt；不然会出现如下编译错误：
undefined reference to `mq_open'
undefined reference to `mq_receive'
undefined reference to `mq_send'
编程实例
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <time.h>
#include <mqueue.h>
#define MSGMAXDATA 512
struct msgbuf {
long mtype;
char mtext[MSGMAXDATA];
};
static void usage(char *prong_name, char *msg)
{
if(msg != NULL)
fputs(msg, stderr);
printf("Usage:%s [options]\n", prong_name);
printf("Options are:\n");
printf("-s send message using msgsnd()\n");
printf("-r send message using msgrcv()\n");
printf("-t message type (default is 1)\n");
printf("-k message queue key (default is 1234)\n");
exit(EXIT_FAILURE);
}
/* system V版本发送消息 */
static void send_msg(int qid, int msgtype)
{
struct msgbuf msg;
time_t t;
msg.mtype = msgtype;
time(&t);
snprintf(msg.mtext, sizeof(msg.mtext), "a message at %s", ctime(&t));
if(msgsnd(qid,(void *)&msg, sizeof(msg.mtext),IPC_NOWAIT) == -1)
{
perror("msgsnd error");
exit(EXIT_FAILURE);
}
printf("sent: %s\n", msg.mtext);
}
/* system V版本接收消息 */
static void get_msg(int qid, int msgtype)
{
struct msgbuf msg;
if(msgrcv(qid, (void *)&msg, sizeof(msg.mtext), msgtype, MSG_NOERROR | IPC_NOWAIT) == -1)
{
if(errno != ENOMSG)
{
perror("msgrcv");
exit(EXIT_FAILURE);
}
printf("No message available for msgrcv()\n");
}
else
printf("message received: %s\n", msg.mtext);
}
/* POSIX 版本发送消息 */
static void send_msg_posix(mqd_t qfd)
{
char msgdata[MSGMAXDATA];
time_t t;
time(&t);
snprintf(msgdata,MSGMAXDATA, "a message at %s", ctime(&t));
if(mq_send(qfd,(void *)msgdata, MSGMAXDATA, 1) == -1)
{
perror("mqsend error");
exit(EXIT_FAILURE);
}
printf("sent: %s\n", msgdata);
}
/* POSIX 版本接收消息 */
static void get_msg_posix(mqd_t qfd)
{
char msgdata[MSGMAXDATA];
int prio = 1;
if(mq_receive(qfd, msgdata, MSGMAXDATA, NULL) == -1)
{
printf("No message available for mq_receive()\n");
printf("errno %d %s\n", errno, strerror(errno));
}
else
{
printf("message received: %s\n", msgdata);
}
}
int main(int argc, char *argv[])
{
int qid, opt;
int mode = 0;
int msgtype = 1;
int msgkey = 1234;
mqd_t mfd = -1;
struct mq_attr attr;
while((opt = getopt(argc, argv, "srSRDt:k:")) != -1)
{
switch(opt)
{
case 's':
mode = 1;
break;
case 'r':
mode = 2;
break;
case 't':
msgtype = atoi(optarg);
if(msgtype <= 0)
usage(argv[0], "-t option must be greater than 0\n");
break;
case 'k':
msgkey = atoi(optarg);
break;
case 'S':
mode = 3;
break;
case 'R':
mode = 4;
break;
case 'D':
mode = 5;
break;
default:
usage(argv[0], "Unrecognized option\n");
}
}
if(mode == 0)
usage(argv[0], "must use either -s,-S or -r,-R option\n");
if(mode == 3 || mode == 4 || mode == 5)
{
attr.mq_maxmsg = 20;
attr.mq_msgsize = MSGMAXDATA;
if((mfd = mq_open("/testfile", O_CREAT | O_RDWR, 0666, &attr)) == -1)
{
printf("create mq fail errno %d %s\n", errno, strerror(errno));
exit(EXIT_FAILURE);
}
mq_getattr(mfd, &attr);
printf("mq_maxmsg:%ld , mq_msgsize:%ld .\n", attr.mq_maxmsg, attr.mq_msgsize);
if(mode == 5)
{
mq_unlink("/testfile");
}
}
qid = msgget(msgkey, IPC_CREAT | 0666);
if(qid == -1)
{
perror("msgget");
exit(EXIT_FAILURE);
}
if(mode == 2)
get_msg(qid, msgtype);
else if(mode == 1)
send_msg(qid, msgtype);
else if(mode == 4)
get_msg_posix(mfd);
else if(mode == 3)
send_msg_posix(mfd);
if(mfd > 0)
mq_close(mfd);
exit(EXIT_SUCCESS);
}
