進程間通訊之消息隊列

進程間通訊之消息隊列

什麼是消息隊列

消息隊列提供了一種從一個進程向另外一個進程發送一個數據塊的方法。  每一個數據塊都被認爲含有一個類型，接收進程能夠獨立地接收含有不一樣類型的數據結構。咱們能夠經過發送消息來避免命名管道的同步和阻塞問題。可是消息隊列與命名管道同樣，每一個數據塊都有一個最大長度的限制。

相關接口函數

• ftok函數

頭文件

#include <sys/types.h>

#include <sys/ipc.h>

函數原型

key_t ftok( const char * fname, int id )

參數

fname就是你指定的文件名（已經存在的文件名）。

id是子序號。雖然是int類型，可是隻使用8bits(1-255）。

返回值

當成功執行的時候，一個key_t值將會被返回，不然 -1 被返回。

• msgrcv/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);

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

參數

msqid：消息隊列的識別碼。

msgp：指向消息緩衝區的指針，此位置用來暫時存儲發送和接收的消息，是一個用戶可定義的通用結構，形態以下

struct msgbuf {

long mtype; /* 消息類型，必須 > 0 */

char mtext[1]; /* 消息文本 */

};

msgsz：消息的大小。

msgtyp：消息類型

msgtyp等於0 則返回隊列的最先的一個消息。

msgtyp大於0，則返回其類型爲mtype的第一個消息。

msgtyp小於0,則返回其類型小於或等於mtype參數的絕對值的最小的一個消息。

msgflg：用來指明核心程序在隊列沒有數據的狀況下所應採起的行動。若是msgflg和常數IPC_NOWAIT合用，則在msgsnd()執行時如果消息隊列已滿，則msgsnd()將不會阻塞，而會當即返回-1，若是執行的是msgrcv()，則在消息隊列呈空時，不作等待立刻返回-1，並設定錯誤碼爲ENOMSG。當msgflg爲0時，msgsnd()及msgrcv()在隊列呈滿或呈空的情形時，採起阻塞等待的處理模式。

返回值

成功執行時，msgsnd()返回0，msgrcv()返回拷貝到mtext數組的實際字節數。失敗二者都返回-1

• msgctl 函數

頭文件

#include <sys/types.h>

#include <sys/ipc.h>

#include <sys/msg.h>

函數原型

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 標識的消息隊列。

返回值

成功返回0，失敗返回-1

實例

工程文件：

comm.h：封裝所需的函數聲明

comm.c：函數的實現

sever.c：服務端

client.c：客戶端

代碼：

comm.h

//comm.h

#pragma once
#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <string.h>
#include <errno.h>

#define _PROJ_NAME_ "./tmp"
#define _PROJ_ID 0x666
#define _SIZE_ 1024

#define _SEVER_TYPE_ 1
#define _CLIENT_TYPE_ 2

struct msgbuf {
    long mtype;       /* message type, must be > 0 */
    char mtext[_SIZE_];    /* message data */
};

int create_msg_queue();
int get_msg_queue();  //for client
int destroy_msg_queue(int msg_id);
int send_msg(int msg_id, int t, const char* msg);
int recv_msg(int msg_id, int t, char* out);

comm.c

//comm.c

#include "comm.h"

//#define _SEVER_TYPE_ 1
//#define _CLIENT_TYPE_ 2


static int comm_msg_queue(int flag)  //for sever
{
    key_t _key = ftok(_PROJ_NAME_, _PROJ_ID);
    if(_key < 0)
    {
        perror("ftok");
        return -1;
    }

    int msg_id = msgget(_key, flag);
    if(msg_id < 0)
    {
        perror("msgget");
        return -2;
    }

    return msg_id; 
}

int create_msg_queue()
{
    int flag = IPC_CREAT | IPC_EXCL | 0644;
    return comm_msg_queue(flag);
}

int get_msg_queue()  //for client
{
    int flag = IPC_CREAT;
    return comm_msg_queue(flag);
}

int destroy_msg_queue(int msg_id)
{
    int ret = msgctl(msg_id, IPC_RMID, NULL);
    if(ret < 0)
    {
        perror("msgctl");
        return -1;
    }
    return 0;
}

int send_msg(int msg_id, int t, const char* msg)
{
    struct msgbuf _buf;
    _buf.mtype = t;
    strncpy(_buf.mtext, msg, strlen(msg)+1);

    int ret = msgsnd(msg_id, &_buf, strlen(_buf.mtext)+1, 0);
    if(ret < 0)
    {
        perror("msgsend");
        return -1;
    }

    return 0;
}

int recv_msg(int msg_id, int t, char* out)
{
    struct msgbuf _buf;
    _buf.mtype = t;

    int ret = msgrcv(msg_id, &_buf, sizeof(_buf.mtext), t, 0);
    if(ret < 0)
    {
        perror("msgrcv");
        return -1;
    }
    strcpy(out, _buf.mtext);

    return 0;
}

sever.c

//sever.c

#include "comm.h"

//#define _SEVER_TYPE_ 1
//#define _CLIENT_TYPE_ 2

int main()
{
    int msg_id = create_msg_queue();
    char buf[_SIZE_];
    
    while(1)
    {
        memset(buf, '\0', sizeof(buf));
        sleep(10);
        recv_msg(msg_id, _CLIENT_TYPE_, buf);
        printf("client -> sever : %s \n", buf);

        if(strncmp("quit", buf, strlen(buf)) == 0)
        {
            break;
        }

        printf("please enter: ");
        fflush(stdout);

        int size = read(0, buf, sizeof(buf)-1);
        buf[size-1] = '\0';
        send_msg(msg_id, _SEVER_TYPE_, buf);

        if(strncmp("quit", buf, strlen(buf)) == 0)
        {
            break;
        }
    }

    destroy_msg_queue(msg_id);

    return 0;
}

client.c

//client.c

#include "comm.h"

//#define _SEVER_TYPE_ 1
//#define _CLIENT_TYPE_ 2

int main()
{
    int msg_id = get_msg_queue();

    char buf[_SIZE_];
    
    while(1)
    {
        printf("please enter: ");
        fflush(stdout);

        int size = read(0, buf, sizeof(buf)-1);
        buf[size-1] = '\0';
        send_msg(msg_id, _CLIENT_TYPE_, buf);

        if(strncmp("quit", buf, strlen(buf)) == 0)
        {
            break;
        }

        memset(buf, '\0', sizeof(buf));

        recv_msg(msg_id, _SEVER_TYPE_, buf);
        printf("sever -> client : %s \n", buf);

        if(strncmp("quit", buf, strlen(buf)) == 0)
        {
            break;
        }

    }

    destroy_msg_queue(msg_id);

    return 0;
}

程序說明：程序運行後服務端會建立一個消息隊列，運行客戶端能夠發送消息給服務端，服務端收到消息後也能夠發送消息給客戶端，客戶端收到消息後又能夠給服務端發消息，以此反覆，直到客戶端或者服務端發送「quit」字符串後，服務端和客戶端均結束程序。