SystemV消息队列来模拟多“客户端”和一个“服务器”的多进程通信

原 作者 畅柯 | 发表于 | 分类于 2017级

Linux消息队列是Linux系统中进程间通信的一种实现机制，POSIX和System V都为Linux提供了实现消息队列的相应API，下面是System V的消息队列的一种应用（其实SystemV消息队列已经不是很不常用了，一般都用POSIX消息队列来实现进程间的通信！但学了还是觉得有必要总结一下。）。

消息队列的存储形式

消息队列在内核中的存在形式是链表，用来存储各个进程发来的消息信息，如图所示（图片来自网络，侵权立删！）：
在这里插入图片描述

消息队列内核数据结构

struct msqid_ds {
	struct ipc_perm msg_perm;     //IPC对象数据结构
	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_t       msg_lrpid;      //最后一个从消息队列中接收数据的进程号
};

消息队列的操作方式

  • 以上信息可以通过以下方式获取：
struct msqid_ds  buf;
msgctl(msgid, IPC_STAT, &buf) ;
  • 关于system v提供的其他消息队列API
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

key_t ftok( const char * fname, int id )
int msgget(key_t key,int msgflg);
int msgsnd(int msqid,const void *msgp,size_t msgsz,int msgflg);
ssize_t msgrcv(int msqid,void *msgp,size_t msgsz,long msgtype,int msgflg);
int msgctl(int msqid,int cmd,struct msqid_ds *buf);

说明：

（1）ftok 系统建立IPC通讯 （消息队列、信号量和共享内存） 时必须指定一个ID值。通常情况下，该id值通过ftok函数得到。
fname 为指定的文件名，id是子序号。虽然是int类型，但是只使用8bits(1-255）。
（2）msgget：用于创建一个新的或打开一个已经存在的消息队列，此消息队列与key相对应。
key：某个消息队列的名字。
msgflg：由9个权限标志构成，它们的用法和创建文件时使用的mode模式标志一样，比如文件中mode模式有O_CREAT，消息队列中有IPC_CREAT等。
（3）msgsnd：向消息队列中添加消息。
msgid：由msgget函数返回的消息队列标识码。
msgp：是一个指针，指针指向准备发送的信息。
msgsz：是msgp指向的消息长度，这个长度不包含保存消息类型的那个long int长整型。
msgflg：控制着当前消息队列满或到达系统上限时将要发生的事情。

msgflg=IPC_NOWAIT表示队列满不等待，返回EAGAIN错误。

（4）msgrcv：从消息队列中获取消息。

msgid：由msgget函数返回的消息队列标识码。
msgp：是一个指针，指向准备接收的消息。
msgsz：是msgp指向的消息长度，这个长度不包含保存消息类型的那个long int长整型。
msgtype：可以实现接收优先级的简单形式。

msgtype=0返回队列第一条消息
msgtype>0返回队列第一条类型等于msgtype的消息
msgtype<0返回队列第一条类型小于等于msgtype绝对值的消息

msgflg：控制着队列中没有相应类型的消息可供接收时将要发生的事。

msgflg=IPC_NOWAIT，队列没有可读消息不等待，返回ENOMSG错误
msgflg=MSS_NOERROR，消息大小超过msgsz时被截断
msgflg>0且msgflg=MSC_EXCEPT，接收类型不等于msgtype的第一条消息

返回值：成功返回实际放到接收缓冲区去的字符个数，失败返回-1

（5）msgctl看到这个，一下明白这是个万能函数，通过它可移除一个消息队列，可以获取消息队列中的消息数等等。

msqid：由msgget函数返回的消息队列标识码。
cmd：将要采取的动作（有三个可取值）。

IPC_STAT 通过这个获取内核消息队列结构中的消息信息。
IPC_STAT
读取消息队列的数据结构msqid_ds，并将其存储在b u f指定的地址中。
IPC_SET
设置消息队列的数据结构msqid_ds中的ipc_perm元素的值。这个值取自buf参数。
IPC_RMID
从系统内核中移走消息队列。

buf：内核消息队列维护的结构。

消息队列应用

客户端

#include <iostream>
#include<sys/ipc.h>
#include<sys/msg.h>
#include<thread>
#include<stdlib.h>
#include<string.h>
#include<sys/types.h>
#include<unistd.h>
using namespace std ;

struct data {
    pid_t pid ;
    char info[1024] ;
} ;

typedef struct msg {
    long type ;
    struct data buf ;
} msg_t;

void sendmsg(int msgid) {
    
    msg_t msg ;
    bzero(&msg, sizeof(msg));
    msg.buf.pid = getpid();;
    cout << msgid << endl;
    while(1) {
        msg.type = 1 ;
        cout << "输入发送的内容：";
        cin >>msg.buf.info ;
        int ret = msgsnd(msgid, (void*)&msg, sizeof(msg_t)-sizeof(long), 0) ;
        if(ret < 0) {
                cout <<errno << endl;
                exit(1) ;
        }
    }
}

void recvmsg(int msgid) {
    
    msg_t msg ;
    msg.type = getpid();
    while(1) {
       while(1){
        int ret = msgrcv(msgid, (void*)&msg, sizeof(msg_t)-sizeof(long), msg.type, 0) ;
           if(ret > 0) {
                break ;
            }
        }
        cout << "\n接收到内容："<< msg.buf.info << endl;
    }
}

int main()
{
    key_t key = ftok(".", 1) ;
    int msgid = msgget(key, 0) ;
    if(msgid < 0) {
        perror("msgget") ;
    }
    cout << msgid << endl;
    thread sendto(sendmsg, msgid) ;
    thread receive(recvmsg, msgid) ;
    sendto.join() ;
    receive.join() ;
    return 0;
}

服务器

#include <iostream>
#include<sys/msg.h>
#include<sys/msg.h>
#include<unistd.h>
#include<thread>
using namespace std ;

struct data {
    pid_t pid ;
    char info[1024] ;
} ;

typedef struct msg {

    long type ;
    struct data buf ;
} msg_t;

void hand(int msgid,int pid);
int main() {
    
    msg_t msg ;
    key_t key = ftok(".", 1) ;
    int msgid = msgget(key, IPC_CREAT|IPC_EXCL|0666) ;
    if(msgid < 0) {
        perror("msgget") ;
    }
    
    while(1) {
        msg.type = 1 ;
        int ret = 0 ;
        while(1){
        ret = msgrcv(msgid, (void*)&msg, sizeof(msg_t) - sizeof(long), msg.type, 0) ;    
            if(ret > 0) {
                break;
            }
        }
        cout << "接收到进程"<<msg.buf.pid <<"的连接:"<< msg.buf.info <<endl ;
        thread t(hand,msgid,msg.buf.pid);
        t.detach();

    }
}   

void hand(int msgid,int pid)
{
    msg_t msg;
    msg.type = pid;
    cout << "请输入发给"<<pid<<"进程"<<"的消息:" ;
    cin>> msg.buf.info ;
    int ret = msgsnd(msgid, (void*)&msg, sizeof(msg_t) - sizeof(long),  0) ;    
    if(ret < 0) {
        perror("msgsnd") ;
        exit(1) ;
    }
}

程序运行的话，先运行“服务器”创建消息队列，再运行多个“客户端”！实现过程中遇到的bug挺多的，太菜了，总是写bug！

  • 运行截图

在这里插入图片描述