阅读更多

1 什么是消息队列(message queue)

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

消息传递的本质：借助内核的帮助，将进程A内存空间的一段数据拷贝到进程B的内存空间当中

2 消息队列的使用

2.1 创建一个消息队列

创建消息队列的函数声明如下：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgget(key_t key, int msgflg);

第一个参数key：一般通过ftok函数来得到（也可以认为是一个端口号）
第二个参数msgflg：与具体的创建方法和权限有关
- 单独的IPC_CREAT表示如果这个消息队列不存在，则创建；存在则打开
- IPC_CREAT | IPC_EXCL表示如果不存在，则创建，存在则出错返回，此举是为了创建一个全新的消息队列

2.2 查看消息队列

消息队列创建好后，可以通过命令行的方式来查看：ipcs -q

2.3 删除消息队列

也可以通过命令行删除它：ipcrm -q key值（或msqid）

当然也可以在程序中删除，要借助msgctl函数：

#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_RMID表示通过msqid来删除
第三个参数buf：用来存储获取自消息队列的一些数据，这里只用到了删除，那么设为NULL即可

2.4 收发消息

下面是msgsnd（发送消息）与msgrcv（接收消息）方法：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

msgp是一个结构体，有些系统实现了，有些需要我们自己定义，例如

struct msgbuf{
    long mtype;
    char mtext[1];
}

mtype用以区分消息类型，因为不同的进程把各自的消息都扔进同一个消息队列，想要取出来自己需要的消息，就需要这个mtype来区分了。mtext是存放数据的数组，大小可以自定义。msgsz就指定了mtext的大小。msgflg用来设置相关的一些模式，这里直接传0，表示以阻塞方式发送（或接收）

3 消息队列的特点

面向数据块的服务
相对于管道而言函数复杂
生命周期随内核（不主动删除不会自动释放）
System V版本的消息队列，由操作系统在底层提供队列支持。消息条数，消息大小，整个系统中的消息队列数有限制

4 C源码

comm.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  
  
extern int ser_type;  
extern int cli_type;  
  
struct msgbuf  
{  
    long mtype;  
    char mtext[_SIZE_];  
};  
  
int create_msg();  
int get_msg();  
int send_msg(int id, char *buf, int type);  
int recv_msg(int id, char *buf_out, int type);  
int destory_msg(int id);

comm.cpp：

#include "comm.h"  
#include <stdio.h>  
#include <stdlib.h>  
#include <string.h>  
  
int cli_type = 1;  
int ser_type = 2;  
  
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.cpp：

#include <stdio.h>  
#include "comm/comm.h"  
#include <unistd.h>  
#include <string.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.cpp：

#include <stdio.h>  
#include "comm/comm.h"  
#include <unistd.h>  
#include <string.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;  
}

5 参考

进程间通信之消息队列

Liuye Notebook

进程通信-消息队列