Linux下进程间通信之消息队列

消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法。 每个数据块都被认为是有一个类型，接收者进程接收的数据块可以有不同的类型值。我们可以通过发送消息来避免命名管道的同步和阻塞问题。消息队列与管道不同的是，消息队列是基于消息的，而管道是基于字节流的，且消息队列的读取不一定是入先出。消息队列与命名管道有一样的不足，就是每个消息的最大长度是有上限的（MSGMAX），每个消息队列的总的字节数是有上限的（MSGMNB），系统上消息队列的总数也有一个上限（MSGMNI）。

下面介绍与消息队列有关的几个函数

1.创建新消息队列或取得已存在消息队列



参数：

key：可以认为是一个端口号，也可以由函数ftok⽣生成。

msgflg：

IPC_CREAT 如果IPC不存在，则创建一个IPC资源，否则打开操作。

IPC_EXCL：只有在共享内存不存在的时候，新的共享内存才建立，否则就产生错误。

如果单独使⽤用IPC_CREAT，XXXget()函数要么返回一个已经存在的共享内存的操作符，要么返回一个新建的共享内存的标识符。如果将IPC_CREAT和IPC_EXCL标志一起使⽤用，XXXget()将返回一个新建的IPC标识符；如果该IPC资源已存在，或者返回-1。

IPC_EXEL标志本身并没有太大的意义，但是和IPC_CREAT标志一起使用可以用来保证得到对象是新建的，而不是打开已有的对象。

2.向队列读/写消息

原型：

msgrcv从队列中取⽤用消息：



参数：

msqid：消息队列的标识码

msgp：指向消息缓冲区的指针，此位置用来暂时存储发送和接收的消息，是一个用户可定义的通用结构。

msgsz：消息的大小。

msgtyp：从消息队列内读取的消息形态。如果值为零，则表示消息队列中的所有消息都会被读取。

msgflg：用来指明核心程序在队列没有数据的情况下所应采取的行动。如果msgflg和常数IPC_NOWAIT合用，则在msgsnd()执⾏行时若是消息队列已满，则msgsnd()将不会阻塞，而会立即返回-1，如果执行的是msgrcv()，则在消息队列呈空时，不做等待马上返回-1，并设定错误码为ENOMSG。当msgflg为0时，msgsnd()及msgrcv()在队列呈满呈空的情形时，采取阻塞等待的处理模式。

3.设置消息队列属性



参数：msgctl 系统调用对 msgqid 标识的消息队列执行 cmd 操作，系统定义了 3 种 cmd 操作： IPC_STAT , IPC_SET , IPC_RMID。

IPC_STAT : 该命令用来获取消息队列对应的 msqid_ds 数据结构，并将其保存到 buf 指定的地址空间。

IPC_SET : 该命令用来设置消息队列的属性，要设置的属性存储在buf中。

IPC_RMID : 从内核中删除 msqid 标识的消息队列。

4.key_t键

ftok函数

函数ftok把一个已存在的路径名和一个整数标识得转换成一个key_t值，称为IPC键：



ftok的典型实现调⽤用stat函数，然后组合以下三个值：

1．pathname所在的文件系统的信息(stat结构的st_dev成员)

2．该⽂文件在本文件系统内的索引节点号(stat结构的st_ino成员)

3. proj_id的低序8位(不能为0)

从程序运行的结果可以看出，ftok调⽤用返回的整数IPC键由proj_id低序8位，st_dev成员的低序8位，st_info的低序16位组合而成。不能保证两个不同的路径名与同一个proj_id的组合产生不同的键，因为上面所列三个条目(文件系统标识符、索引节点、proj_id)中的信息位数可能大于一个整数的信息位数。

下面用代码编写程序实现一下消息队列：

comm.h

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define _PATH_NAME_ "./comm/comm.h"
#define _PROJ_ID_ 0x111
#define _SIZE_ 1024

int cli_type = 1;
int ser_type = 2;

struct msgbuf
{
long mtype;
char mtext[_SIZE_];
};
static int comm_msg(int flag)
{
int key = ftok(_PATH_NAME_, _PROJ_ID_);
if (key < 0)
{
perror("ftok");
exit(1);
}

int id = msgget(key, flag);
if (id < 0)
{
perror("msgget");
exit(2);
}

return id;
}

int create_msg()
{
int flag = IPC_CREAT | IPC_EXCL | 0666;
return comm_msg(flag);
}

int get_msg()
{
int flag = IPC_CREAT;
return comm_msg(flag);
}

int send_msg(int id, char *buf, int type)
{
struct msgbuf msg;
msg.mtype = type;
strncpy(msg.mtext, buf, strlen(buf)+1);

int ret = msgsnd(id, &msg, sizeof(msg.mtext), 0);   // 0缺省阻塞
if (ret < 0)
{
perror("msgsnd");
return ret;
}
return 0;
}

int recv_msg(int id, char *buf_out, int type)
{
struct msgbuf msg;
size_t _s = msgrcv(id, &msg, sizeof(msg.mtext), type, 0);
if (_s < 0)
{
perror("msgrcv");
return _s;
}
strcpy(buf_out, msg.mtext);
}

int destory_msg(int id)
{
msgctl(id, IPC_RMID, NULL);
}

server.c

#include "comm.h"

int main()
{
int id = create_msg();
printf("%d\n", id);

char buf[_SIZE_];
while (1)
{
memset(buf, 0, sizeof(buf));

recv_msg(id, buf, cli_type);
printf("client# %s\n", buf);
if (strcasecmp(buf, "quit") == 0)
break;

memset(buf, 0, sizeof(buf));
printf("Please Enter# ");
fflush(stdout);
ssize_t _s = read(0, buf, sizeof(buf));
if (_s > 0)
{
buf[_s - 1] = 0;
}
send_msg(id, buf, ser_type);
}

destory_msg(id);

return 0;
}

client

#include "comm.h"

int main()
{
int id = get_msg();
printf("%d\n", id);

char buf[_SIZE_];
while (1)
{
memset(buf, 0, sizeof(buf));
printf("Please Enter# ");
fflush(stdout);
ssize_t _s = read(0, buf, sizeof(buf));
if (_s > 0)
{
buf[_s - 1] = 0;
}
send_msg(id, buf, cli_type);
if (strcasecmp(buf, "quit") == 0)
break;

memset(buf, 0, sizeof(buf));
recv_msg(id, buf, ser_type);
printf("server# %s\n", buf);

}

return 0;
}