linux 进程间通信之 消息队列

消息队列就是一个消息的链表。

能够把消息看作一个记录，具有特定的格式以及特定的优先级。对消息队列有写权限的进程能够向中依照一定的规则加入新消息。有读权限的进程则能够读走消息。

读走就没有了。消息队列是随内核持续的。 仅仅有内核重新启动或人工删除时。该消息才会被删除。

在系统范围内，消息队列与键值唯一相应。

关于消息队列使用的API

key_t ftok(const char *pathname, int proj_id);
#在IPC中，我们经经常使用一个 key_t 的值来创建或者打开 信号量，共享内存和消息队列。

这个 key_t 就是由ftok函数产生的。

pathname：指定的文件名称，该文件必须是存在并且能够訪问
proj_id：1~255之间的整数值。
对于ftok这个函数，个人是认为没有太大用处，第一，应用程序可能会在不同的主机上使用，（换了一个环境，文件有没有？）。第二点，假设在訪问同一共享内存的多个进程先后调用ftok这个时间段中， pathname指定的文件被删除且又一次创建，那么，每一个进程得到的 key_t 是不一样的，应用程序不会报错，可是数据共享的目的是达不到了。


int msgget(key_t key, int msgflg);
函数功能：打开、创建消息队列
key： 使用 IPC_PRIVATE 意味着即将创建新的消息队列。 官方的说法是用ftok产生一个key（鉴于上面关于ftok的介绍。个人比較认同的做法是自定义一个）使用。
msgflg：是一组标志。
IPC_CREAT：假设消息队列不存在，则创建一个消息队列。

IPC_EXCL：仅仅有在消息队列不存在的时候，新的消息队列才建立。否则就产生错误。
对于这个參数通常是这样操作
#define PERM S_IRUSR | S_IWUSR | IPC_CREAT
然后把 PERM 当作msgflg。

成功返回消息队列的标识符。不成功返回-1，并设置errno。

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
函数功能：向消息队列发送一条消息
msqid： 为消息队列的id（msgget函数的返回值）
msgp：存放消息的结构体。一般例如以下定义：
struct msgbuf{
long m_type; // 消息类型，必须大于0
char m_text[n]; //消息的内容
}
msgsz：消息的长度（也就是msgbuf.m_text的长度）
msgflg：參考函数 msgrcv 。
成功返回0。否则返回-1。


ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
函数功能：从消息队列中读取消息
msqid： 为消息队列的id（msgget函数的返回值）
msgp：存放消息的结构体。一般例如以下定义：
struct msgbuf{
long m_type; // 消息类型，必须大于0
char m_text[n]; //消息的内容
}
msgsz：消息的长度（也就是msgbuf.m_text的长度）
msgtyp：请求读取的消息类型，取值为
0 : 表示读取消息队列中的第一条消息
>0: 表示队列中类型为 msgtyp 的第一条消息被读取；假设 msgflg 设置成 MSG_EXCEPT ，则表示消息队列中除了类型是 msgtyp 的第一条消息将被读取。

msgflg：是一组标志。读消息标志msgflg能够为下面几个常值的或
IPC_NOWAIT 假设没有满足条件的消息，调用马上返回，此时，errno=ENOMSG（设置了就是不堵塞。否则就是堵塞）
MSG_EXCEPT 与msgtyp>0配合使用，返回队列中第一个类型不为msgtyp的消息
MSG_NOERROR 假设队列中满足条件的消息内容大于所请求的msgsz字节，则把该消息截断。截断部分将丢失。

成功返回读出消息的实际字节数。否则返回-1。

int msgctl(int msqid, int cmd, struct msqid_ds *buf);
函数功能：控制消息队列的使用
msqid：为消息队列的id （msget函数的返回值）
cmd：控制命令。

可取值例如以下：
IPC_STAT：得到消息队列的状态
IPC_SET：改变消息队列的状态
IPC_RMID：删除消息队列
buf：是一个结构体指针，当cmd为IPC_STAT的时候。取得的状态放入这个结构体中。假设要改变消息队列的状态，用这个结构体指定。
struct msqid_ds 结构体原型
struct msqid_ds {
    struct ipc_perm msg_perm;     /* 操作权限 */
    time_t          msg_stime;    /* 最后一个进程发送消息到消息队列的时间 */
    time_t          msg_rtime;    /* 最后一个进程读取消息队列中消息的时间 */
    time_t          msg_ctime;    /* 最后一个进程改动消息队列的时间 */
    unsigned long   __msg_cbytes; /*  当前在队列的字节数(标准) */
    msgqnum_t       msg_qnum;     /*  当前在队列的消息数量 */
    msglen_t        msg_qbytes;   /* 队列中同意的最大字节数 */
    pid_t           msg_lspid;    /* 最后一个发送消息的进程pid */
    pid_t           msg_lrpid;    /* 最后一个接收消息的进程pid */
};
成功返回0，否则返回-1。

演示样例程序

my_head.h

#ifndef MY_HEAD_H_INCLUDED
#define MY_HEAD_H_INCLUDED


/*
包括经常使用头文件。编写測试代码用
*/

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/msg.h>
#include <sys/stat.h>
#include <errno.h>


#endif // MY_HEAD_H_INCLUDED

msg.h

#include "my_head.h"



#define PERM    S_IRUSR | S_IWUSR | IPC_CREAT

struct msgbuf{
    long m_type;
    char m_text[128];
};


int open_msg_queue(key_t keyval);

int send_msg_to_queue(int msqid, struct msgbuf buf, size_t msgsz);

int recv_msg_from_queue(int msqid, struct msgbuf *buf, size_t msgsz);

int delete_msg_queue(int msqid);

msg.c

#include "msg.h"




int open_msg_queue(key_t keyval){
    int qid=msgget(keyval,PERM);
    if(qid == -1)
        fprintf(stderr,"%s %s %d
",strerror(errno),__FILE__,__LINE__);

    return qid;
}

int send_msg_to_queue(int msqid, struct msgbuf buf, size_t msgsz){
    if(msgsnd(msqid,&buf,msgsz,0) == -1){
        fprintf(stderr,"%s %s %d
",strerror(errno),__FILE__,__LINE__);
        return -1;
    }

    return 0;
}

int recv_msg_from_queue(int msqid, struct msgbuf *buf, size_t msgsz){
    if(msgrcv(msqid,buf,msgsz,0,0) == -1){
        fprintf(stderr,"%s %s %d
",strerror(errno),__FILE__,__LINE__);
        return -1;
    }
    return 0;
}

int delete_msg_queue(int msqid){
    if(msgctl(msqid,IPC_RMID,NULL) == -1){
        fprintf(stderr,"%s %s %d
",strerror(errno),__FILE__,__LINE__);
        return -1;
    }
    return 0;
}

msg_send.c

#include "msg.h"




int main(){
    struct msgbuf msg_buf;

    int msqid = open_msg_queue((key_t)1234);

    while(1){
        printf("Enter: ");
        fgets(msg_buf.m_text,sizeof(msg_buf.m_text),stdin);
        
        send_msg_to_queue(msqid,msg_buf,sizeof(msg_buf.m_text));
        
        if(strncmp(msg_buf.m_text,"quit",4) == 0)
            break;
    }






    return 0;
}

msg_recv.c

#include "msg.h"





int main(){
    struct msgbuf msg_buf;

    int msqid = open_msg_queue((key_t)1234);

    while(1){

        recv_msg_from_queue(msqid,&msg_buf,sizeof(msg_buf.m_text));
        
        printf("recv: %s",msg_buf.m_text);


        if(strncmp(msg_buf.m_text,"quit",4) == 0)
            break;
    }

    delete_msg_queue(msqid);




    return 0;
}