细说linux IPC（十）：system V 消息队列
浏览数：59 /
时间：2015年06月20日
【版权声明：尊重原创，转载请保留出处：blog.csdn.net/shallnet 或 .../gentleliu，文章仅供学习交流，请勿用于商业用途】
system V消息队列和posix消息队列类似，linux系统这两种消息队列都支持。先来看一下system V消息队列相关操作及其函数。
- msgget()函数创建一个消息队列或打开一个消息队列。
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);参数key为ftok返回值或IPC_PRIVATE；
参数msgflg为IPC_CREAT或和IPC_CREAT | IPC_EXCL，当消息队列已经存在时，而msgflg指定为IPC_CREAT | IPC_EXCL，函数返回EEXIST。
在系统中的每一个消息队列，内核维护这样一个结构体：
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 (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) */
};其中ipc_perm 结构体定义如下：
struct ipc_perm {
key_t __key; /* Key supplied to msgget(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 */
};当一个消息队列初始化完后，该消息队列对应的结构被作如下初始化：
msg_perm.cuid 和 msg_perm.uid被设置为当前进程的用户ID，msg_perm.cuid 和 msg_perm.uid被设置为当前进程的组ID；msg_perm.mode被设置为msgflg；msg_qnum, msg_lspid, msg_lrpid, msg_stime and msg_rtime
被设置为 0；msg_ctime被设置为当前时间；msg_qbytes被设置为系统限制值MSGMNB。
- msgsnd()函数向消息队列中添加消息。
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);参数msgqid为消息队列标识，为msgget()返回值；
参数msgp为指向用户定义结构体的指针，结构体定义格式模板如下：
struct msgbuf {
long mtype; /* message type, must be > 0 */
char mtext[1]; /* message data */
}; mtex域为数组，该数组大小为函数参数msgsz，mtype的值必须大于0；该结构由用户自己定义，一般根据实际应用来定义。
参数msgsz为将要发送消息的长度，可以指定为0，一般指定为msgbuf结构体长度减去long类型长度（sizeof(msgbuf) - sizeof(long)）；
参数msgflg为0，或IPC_NOWAIT，IPC_NOWAIT为非阻塞，发送时如果对列数据已满，则函数马上返回错误EAGAIN，如果msgflg为0，则阻塞直到消息队列由足够空间添加新的消息，或消息队列从系统中删除，或被信号中断。
- msgrcv()函数从消息队列中读取消息。
#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);参数msgid同函数msgsnd；
参数msgp通函数函数msgsnd，为指向接收消息结构体指针；
参数msgsz为接受消息的缓存大小；
参数msgtyp为读取何种消息，如果为：
0 - 读取消息队列中最早的消息，以先进先出的方式读取消息。
> 0 - 读取消息队列类型为msgtyp最早的消息。
< 0 - 读取消息队列中，消息类型小于或等于msgtyp绝对值的消息中最小消息类型的最早消息。
参数msgflg为如下的值或者几个值得或：
IPC_NOWAIT - 如果队列中没有则立即返回；如果未指定该选项，函数将一直阻塞，直到指定的消息可读取，或消息队列从系统中删除，或被信号中断；
MSG_NOERROR - 截断消息如果接收到的消息大小大于msgsz给定的值；
MSG_EXCEPT - 使用大于0的msgtyp去读取消息类型不为msgtyp第一个消息；
msgctl()函数对一个队列做控制操作。
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
参数cmd为对队列要做的操作，这些操作有：
IPC_STAT - 获取消息队列信息，从内核数据结构中拷贝消息队列为msqid的结构体到buf指针指向的msqid_ds 结构体中。
IPC_STAT - 获取消息队列信息，从内核数据结构中拷贝消息队列为msqid的结构体到buf指针指向的msqid_ds 结构体中。
IPC_SET - 使用buf指向的结构体来设置msgqid对应的消息队列内核结构体；
IPC_RMID - 删除msgqid消息队列；
IPC_INFO，MSG_INFO， MSG_STAT详见linux man函数手册。
示例：
服务进程：
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <errno.h>
#include <unistd.h>
#include <signal.h>
#include <stdlib.h>
#include "slnmq.h"
static int recv_type, msgid;
int sln_server_loop(void)
{
int rcvmsg_len;
key_t key;
struct sln_msgbuf slnmsg;
key = ftok(SLN_IPC_MQ_NAME, 0);
if (key < 0) {
fprintf(stderr, "ftok: %s\n", strerror(errno));
return -1;
}
msgid = msgget(key, IPC_CREAT | IPC_EXCL);
if ((msgid < 0) && (errno != EEXIST)) {
fprintf(stderr, "msgget: %s\n", strerror(errno));
return -1;
}
if (msgid < 0) {
printf("Open existing mq!\n");
msgid = msgget(key, IPC_CREAT);
if (msgid < 0) {
fprintf(stderr, "msgget: %s\n", strerror(errno));
return -1;
}
}
sleep(10); //just for test
for (;;) {
rcvmsg_len = msgrcv(msgid, &slnmsg, SLN_IPC_MQ_MSGSIZE, recv_type, 0);
if (rcvmsg_len < 0) {
fprintf(stderr, "msgrcv: %s\n", strerror(errno));
sleep(1);
continue;
}
fprintf(stdout, "receive - recv len: %d, msg type: %d, msg: %s\n",
rcvmsg_len, slnmsg.msgtype, slnmsg.msg);
}
return 0;
}
static void sigint_func(int sig)
{
if (msgctl(msgid, IPC_RMID, NULL) < 0) {
fprintf(stderr, "msgctl: %s\n", strerror(errno));
}
exit(0);
}
int main(int argc, const char *argv[])
{
FILE *fp = NULL;
if (argc != 2) {
fprintf(stderr, "Usage: %s <msg type>\n", argv[0]);//运行时，指定服务要进程接收的消息类型
return -1;
}
recv_type = atoi(argv[1]);
if (access(SLN_IPC_MQ_NAME, F_OK) < 0) {
fp = fopen(SLN_IPC_MQ_NAME, "w+");
if (NULL != fp) {
fclose(fp);
}
}
signal(SIGINT, sigint_func); //服务进程捕获中断信号，在中断处理函数里面删除消息队列并推出程序
sln_server_loop();
return 0;
}客户进程：
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <errno.h>
#include "slnmq.h"
int sln_msgsnd(int msg_type, const void *sndmsg, int sndlen)
{
int msgid, rcvmsg_len;
key_t key;
struct sln_msgbuf slnmsg;
key = ftok(SLN_IPC_MQ_NAME, 0);
if (key < 0) {
fprintf(stderr, "ftok: %s\n", strerror(errno));
return -1;
}
msgid = msgget(key, IPC_CREAT);
if (msgid < 0) {
fprintf(stderr, "msgget: %s\n", strerror(errno));
return -1;
}
slnmsg.msgtype = msg_type;
memcpy(&slnmsg.msg, sndmsg, sndlen);
msgsnd(msgid, (void *)&slnmsg, sndlen, 0);
}
int main(int argc, const char *argv[])
{
int type;
if (argc != 3) {
fprintf(stderr, "Usage: %s <msg type> <msg content>\n", argv[0]);
return -1;
}
type = atoi(argv[1]);
sln_msgsnd(type, argv[2], strlen(argv[2]) + 1);
return 0;
}服务进程首先运行，并在开始读取消息处等待10秒，客户进程在服务进程还未开始读取数据时向消息队列加入多条消息，如下：
./client 1 abcd ./client 2 efghi ./client 3 jkl ./client 4 mn ./client 5 opqrst ./client 6 uv ./client 7 w ./client 8 xyz
当服务进程运行时会从消息队列中取出服务进程要取出类型（运行时参数指定）的消息，下面分别是服务器运行时指定的消息类型以及取出的消息（客户端的运行都如上）：
# ./server 0（读取所有消息） receive - recv len: 5, msg type: 1, msg: abcd receive - recv len: 6, msg type: 2, msg: efghi receive - recv len: 4, msg type: 3, msg: jkl receive - recv len: 3, msg type: 4, msg: mn receive - recv len: 7, msg type: 5, msg: opqrst receive - recv len: 3, msg type: 6, msg: uv receive - recv len: 2, msg type: 7, msg: w receive - recv len: 4, msg type: 8, msg: xyz
# ./server 4 （读取消息类型为4的消息） Open existing mq! receive - recv len: 3, msg type: 4, msg: mn
# ./server -5 （读取消息小于5的消息） Open existing mq! receive - recv len: 5, msg type: 1, msg: abcd receive - recv len: 6, msg type: 2, msg: efghi receive - recv len: 4, msg type: 3, msg: jkl receive - recv len: 3, msg type: 4, msg: mn receive - recv len: 7, msg type: 5, msg: opqrst
在服务器未退出时，可以使用系统命令ipcs查看系统中的消息队列：
# ipcs ------ Message Queues -------- key msqid owner perms used-bytes messages 0x001fffde 229376 root 0 0 0
当用户按下CTRL+C时，程序会删除掉该消息队列，按下CTRL+C后，再查看：
# ipcs ------ Message Queues -------- key msqid owner perms used-bytes messages
可以看到程序中的删除是成功的。
posix消息队列和system V消息队列大致类似，不过通过这两节可以看到有一些差别：
posix消息队列有优先级的概念，system V没有，只能按照消息类型的先进先出来读取消息即FIFO；
system V消息队列有消息类型的概念，并且可以按照消息类型取出消息，而posix不能区分消息类型；
本节源码下载：
郑重声明：本站内容如果来自互联网及其他传播媒体，其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享，并不代表本站赞同其观点和对其真实性负责，也不构成任何其他建议。
