Linux中的消息队列

半只蚂蚁

　　消息队列是基于消息的，而管道是基于字节流的，且消息队列的读取不一定是先进先出。消息队列与命名管道有一样的不足，就是每个消息的最大长度是有上限的（ MSGMAX），每个消息队列的总的字节数是有上限的（ MSGMNB），系统上消息队列的总数也有一个上限（ MSGMNI）。

　　消息队列是用链表实现的。

　　1.创建新消息队列或取得已存在消息队列
　　原型：
int msgget(key_t key, int msgflg);　　参数：
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从队列中取用消息：
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);　　msgsnd将数据放到消息队列中：
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);　　参数：
msqid：消息队列的标识码
msgp：指向消息缓冲区的指针，此位置用来暂时存储发送和接收的消息，是一个用户可定义的通用结构，形态如下：

struct msgstru{
long mtype; //于0
char mtext[用户指定大小];
};　　

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

　　3.设置消息队列属性
　　原型：
int msgctl ( int msgqid, int cmd, struct msqid_ds *buf );　　参数： msgctl 系统调用对 msgqid 标识的消息队列执行cmd 操作，系统定义了 3 种 cmd 操作： IPC_STAT , IPC_SET , IPC_RMID
IPC_STAT : 该命令用来获取消息队列对应的 msqid_ds 数据结构，并将其保存到 buf 指定的地址空间。
IPC_SET : 该命令用来设置消息队列的属性，要设置的属性存储在buf中。
IPC_RMID : 从内核中删除 msqid 标识的消息队列。

　　 key_t键
System V IPC使key_t值作为它们的名字，在Redhat linux(后续验证默认都在该平台下)下
key_t被定义为int类型，追溯如下：

/usr/include/sys/ipc.h
#ifndef __key_t_defined
typedef __key_t key_t;
#define __key_t_defined
#endif
/usr/include/bits/types.h
typedef __DADDR_T_TYPE __daddr_t; /* The type of a disk address. */
typedef __SWBLK_T_TYPE __swblk_t; /* Type of a swap block maybe? */
typedef __KEY_T_TYPE __key_t; /* Type of an IPC key */
/usr/include/bits/typesizes.h
#define __KEY_T_TYPE __S32_TYPE
/usr/include/bits/types.h
#define __S32_TYPE int　　ftok函数
　　函数ftok把一个已存在的路径名和一个整数标识得转换成一个key_t值，称为IPC键：

# include <sys/types.h>
# include <sys/ipc.h>
key_t ftok(const char *pathname, int proj_id);　　代码：

#include "unpipc.h"
int main(int argc, char **argv)
{
struct stat stat;
if (argc != 2)
err_quit("usage: ftok ");
Stat(argv[1], &stat);
printf("st_dev: %lx, st_ino: %lx, key: %x/n",
(u_long)stat.st_dev, (u_long)stat.st_ino,
Ftok(argv[1], 0x57));
exit(0);
}　　运行结果：

[cbs@linux svipc]$ ./ftok /tmp/mysql.sock
st_dev: 802, st_ino: 34219, key: 57024219　　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位组合而成。

　　消息队列的实现代码：

　　common.h
#ifndef __COMM__
#define __COMM__
#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/types.h>
#include <string.h>
#include <time.h>
#define __MSG_SIZE__ 1024
#define FILEPATH "/tmp/.msg"
#define ID 0
extern const int g_ser_send_type;//server
extern const int g_cli_send_type;//client
typedef struct _msginfo{
long mtype;
char mtext[__MSG_SIZE__];
}msginfo;
void print_log(char *);
#endif　　common.c
#include "common.h"
const int g_ser_send_type = 1;//server
const int g_cli_send_type = 2;//client
void print_log(char *msg)
{
printf("%s[%d] : %s\n", __FUNCTION__, __LINE__, msg);
}　　msg_server.h

#ifndef _MSG_SERVER_
#define _MSG_SERVER_
#include"common.h"
int msg_server_start();
int msg_server_end(int);
#endif　　msg_server.c

#include "msg_server.h"
int _msg_id = -1;
int msg_server_start()
{
key_t _key = ftok(FILEPATH, ID);//创建键值
if (_key < 0){
print_log("get key id error");
return 1;
}
msginfo _ser_info;
_msg_id = msgget(_key, IPC_CREAT); //获取信号队列ID
if (_msg_id < 0){
print_log("msg_server get key id failed\n");
return 1;
}
while (1){
// sleep(10);
if (msgrcv(_msg_id, &_ser_info, sizeof(_ser_info), g_cli_send_type, 0) == -1){
print_log("msg rcv error");
return 1;
}
printf("client :> %s\n", _ser_info.mtext);
printf("server :>");
memset(_ser_info.mtext, '\0', sizeof(_ser_info.mtext));
fgets(_ser_info.mtext, __MSG_SIZE__, stdin);
if (strncasecmp(_ser_info.mtext, "quit", 4) == 0){
printf("server bye!\n");
break;
}
_ser_info.mtype = g_ser_send_type;
if (msgsnd(_msg_id, &_ser_info, __MSG_SIZE__, 0) == -1){
printf("server send msg error\n");
exit(0);
}
}
return 0;
}
int msg_server_end(int id)
{
if (msgctl(id, IPC_RMID, NULL) == -1){
printf("delete msg kernel info error\n");
return 1;
}
return 0;
}
static void delete_msg(void)
{
if (_msg_id != -1){
msg_server_end(_msg_id);
}
printf("delete msg queue end\n");
}
int main(int argc, char *argv[])
{
atexit(delete_msg);
if (msg_server_start() == 0){
print_log("msg_server start success\n");
}
else{
print_log("msg_server start failed\n");
}
return 0;
}　　msg_client.h

#ifndef _MSG_CLIENT_
#define _MSG_CLIENT_
#include"comm.h"
int msg_client_start();
int msg_client_end(int);
#endif　　msg_client.c
#include "msg_client.h"
int _msg_id = -1;
int msg_client_start()
{
key_t _key = ftok(FILEPATH, ID);//创建键值
if (_key < 0){
print_log("client get key id error");
return 1;
}
msginfo _cli_info;
_msg_id = msgget(_key, 0); //获取信号队列ID
if (_msg_id < 0){
print_log("msg_server get key id failed\n");
return 1;
}
while (1){
printf("client :>");
fgets(_cli_info.mtext, sizeof(_cli_info.mtext), stdin);
if (strncasecmp(_cli_info.mtext, "quit", 4) == 0){
printf("client bye!\n");
break;
}
_cli_info.mtype = g_cli_send_type;
if (msgsnd(_msg_id, &_cli_info, sizeof(_cli_info), 0) == -1){
printf("client send msg error\n");
exit(0);
}
memset(_cli_info.mtext, '\0', sizeof(_cli_info.mtext));
if (msgrcv(_msg_id, &_cli_info, __MSG_SIZE__, g_ser_send_type, 0) == -1){
print_log("client recive msg error");
return 1;
}
printf("server :>%s\n", _cli_info.mtext);
}
return 0;
}
int msg_client_end(int id)
{
if (msgctl(id, IPC_RMID, NULL) == -1){
return 1;
}
return 0;
}
static void delete_msg(void)
{
if (_msg_id != -1){
msg_client_end(_msg_id);
}
printf("delete msg queue end\n");
}
int main(int argc, char *argv[])
{
atexit(delete_msg);
if (msg_client_start() == 0){
print_log("msg_server start success\n");
}
else{
print_log("msg_server start failed\n");
}
return 0;
}