程序間通訊-訊息佇列

阿新 • • 發佈：2020-12-30

場景：同一臺linux裝置下，多個程序之間資料要互相通訊，就可以使用訊息佇列進行資料傳輸。

訊息佇列（Message Queue）：是訊息的連結串列，存放在核心中並由訊息佇列識別符號標識

優點：可以實現任意程序間的通訊，並通過系統呼叫函式來實現訊息傳送和接收之間的同步，無需考慮同步問題，方便
缺點：資訊的複製需要額外消耗 CPU 的時間，不適宜於資訊量大或操作頻繁的場合

訊息佇列提供了一個從一個程序向另一個程序傳送資料塊的方法，每個資料塊都可以被認為是有一個型別，接收者接受的資料塊可以有不同的型別。

訊息佇列函式：

1、msgget
功能：建立和訪問一個訊息佇列
原型：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflag);
引數：
key：訊息佇列名稱，用ftok()產生
msgflag:IPC_CREAT或者IPC_EXCL，如果訊息佇列不存在則建立之，如果存在則開啟返回；單獨使用IPC_EXCL是沒有意義的；兩個同時使用，如果訊息佇列不存在則建立之，如果存在則出錯返回。
返回值：成功返回非負整數，即訊息佇列的標識碼，失敗返回-1

2、ftok
功能：建立訊息佇列名稱
原型：
#include <sys/types.h>
#include <sys/ipc.h>
key_t ftok(const char *pathname, int proj_id);
返回值：成功返回一個key值，用於建立訊息佇列，失敗返回-1

3、msgctl
功能：訊息佇列的控制函式
原型：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
引數：
mgqid：由msgget函式返回的訊息佇列標識
cmd：命令，這邊有三個命令
IPC_STAT把msqid_ds結構中的資料設定為訊息佇列的當前關聯值
IPC_SET在程序有足夠許可權的前提下，把訊息佇列的當前值設定為msqid_ds資料結構中給出的值
IPC_RMID刪除訊息佇列
返回值：成功返回0，失敗返回-1

4、msgsnd
功能：把一條訊息新增到訊息佇列中
原型：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
引數：
msgid：msgget函式返回的訊息佇列標識碼
msgp：指標指向準備傳送的訊息
msgsz：msgp指向的訊息的長度
msgflg：預設為0
返回值：成功返回0，失敗返回-1
傳送的訊息結構：
struct msgbuf
{
     long mtye;
     char mtext[1];
};

5、msgrcv
功能：從一個訊息佇列接收訊息
原型：
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
引數：
msgid：msgget函式返回的訊息佇列標識碼
msgp：指標指向準備傳送的訊息
msgsz：msgp指向的訊息的長度
msgtyp：訊息型別
type==0，返回佇列中第一個訊息
type > 0, 返回佇列中訊息型別為type的第一個訊息
type < 0, 返回佇列中訊息型別值小於或等於type絕對值的訊息，如果這種訊息有若干個，則取型別值最小的訊息
msgflg：預設為0

上原始碼：

//comm.h
#ifndef _COMM_H_
#define _COMM_H_
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <sys/msg.h>
struct msgbuf
{
    long mtype;
    char mtext[1024];
};
#define SERVER_TYPE 1
#define CLIENT_TYPE 2
int createMsgQueue();
int getMsgQueue();
int destoryMsgQueue(int msg_id);
int sendMsgQueue(int msg_id, int who, char *msg);
int recvMsgQueue(int msg_id, int recvType, char out[]);
#endif

//comm.c
#include "comm.h"
static int commMsgQueue(int cmd)
{
    key_t key = ftok("/tmp", 0x6666);
    if(key < 0)
    {
        perror("ftok");
        return -1;
    }
    int msg_id = msgget(key, cmd);
    if(msg_id < 0)
    {
        perror("msgget");
    }
    return msg_id;
}
int createMsgQueue()
{
    return commMsgQueue(IPC_CREAT|IPC_EXCL|0666);
}
int getMsgQueue()
{
    return commMsgQueue(IPC_CREAT);
}
int destoryMsgQueue(int msg_id)
{
    if(msgctl(msg_id, IPC_RMID, NULL) < 0)
    {
        perror("msgctl");
        return -1;
    }
    return 0;
}
int sendMsgQueue(int msg_id, int type, char *msg)
{
    struct msgbuf buf;
    buf.mtype = type;
    strcpy(buf.mtext, msg);
    if(msgsnd(msg_id, (void*)&buf, sizeof(buf.mtext), 0) < 0)
    {
        perror("msgsnd");
        return -1;
    }
    return 0;
}
int recvMsgQueue(int msg_id, int recvType, char out[])
{
    struct msgbuf buf;
    int size = sizeof(buf.mtext);
    if(msgrcv(msg_id, (void *)&buf, size, recvType, 0) < 0)
    {
        perror("msgrcv");
        return -1;
    }
    strncpy(out, buf.mtext, size);
    out[size] = 0;
    return 0;
}

//server.c
#include "comm.h"
int main()
{
    int msgid = createMsgQueue();
    printf("msgid = %d\n", msgid);
    char buf[1024] = {0};
    while(1)
    {
        recvMsgQueue(msgid, CLIENT_TYPE, buf);
        if(strcasecmp("quit", buf) == 0)
            break;
        printf("client:%s\n", buf);
        /*
        printf("Please enter:");
        fflush(stdout);
        ssize_t s = read(0, buf, sizeof(buf));
        if(s > 0)
        {
            buf[s - 1] = 0;
            sendMsgQueue(msgid, SERVER_TYPE, buf);
            printf("send done, wait recv...\n");
        }
        */
    }
    destoryMsgQueue(msgid);
    return 0;
}

//client.c
#include "comm.h"
int main()
{
    int msgid = getMsgQueue();
    printf("msgid:%d\n", msgid);
    char buf[1024] = {0};
    while(1)
    {
        printf("Please Enter:");
        fflush(stdout);
        ssize_t s = read(0, buf, sizeof(buf));
        if(s > 0)
        {
            buf[s - 1] = 0;
            sendMsgQueue(msgid, CLIENT_TYPE, buf);
            if(strcasecmp("quit", buf) == 0)
                break;
            //printf("send done, wait recv...\n");
        }
        //recvMsgQueue(msgid, SERVER_TYPE, buf);
        //printf("server#%s\n", buf);
    }
    return 0;
}

編譯執行client、server。

client傳送訊息：

[[email protected] tmp]# ./client

msgid:360448

Please Enter:AAAA

Please Enter:BBBBB

Please Enter:HEIIL

Please Enter:quit

[[email protected] tmp]#

server接收訊息：

[[email protected] tmp]# ./server

msgid = 360448