學習環境 Centos6.5 Linux 核心 2.6

什麼是訊息佇列？

訊息佇列是SystemV版本中三種程序通訊機制之一，另外兩種是訊號量和共享儲存段。訊息佇列提供了程序間傳送資料塊的方法，而且每個資料塊都有一個型別標識。訊息佇列是基於訊息的，而管道是基於位元組流。建立的訊息佇列，生命週期隨核心，只有核心重啟或使用者主動去刪除，才可以真正關閉訊息佇列。

背景知識：

• I P C 識別符號：每一個核心中的IPC結構（訊息佇列，訊號量，共享儲存段）都用一個非負整數的識別符號（identifier）加以引用。當一個訊息佇列傳送或取訊息，只需要知道其佇列標示符。

// 核心為每個IPC物件維護一個數據結構（/usr/include/linux/ipc.h）
 

struct ipc_perm
{
    key_t __key; /* key supplied to xxxget(2) */
    uid_t uid;  /* Effective UID of owner */
    gid_t gid;  /* Effective GID of owner */
    uid_t cuid; /* Effective UID of creator */
    gid_t cgid; /* Effective GID of creator */
    unsigned short mode;  /* Permission */
    unsigned short
 
 __seq; /* Sequeence number*/
}

• IPC關鍵字：因為IPC識別符號是IPC結構的內部名。為使多個合作程序能夠在同一IPC物件上會合，需要提供一個外部名方案。即鍵（key）每一個IPC物件都與一個鍵相關聯，於是鍵就作為該結構的外部名。要想獲得一個唯一識別符號，必須使用一個IPC關鍵字。server和client程序必須雙方都同意此關鍵字。 可以使用ftok( )函式為客戶端和伺服器產生關鍵字值。

//訊息佇列的結構 ( /usr/include/linux/msg.h)
// message queue id
// defined in <linux/ipc.h>
struct
 
 msqid_ds
{
    struct ipc_perm msg_perm;
    struct msg* msg_first;      /* first message on queue, unused */
    struct msg* msg_last;       /* last message in queue, unused */
    __kernel_time_t msg_stime;  /* last msgsnd time */
    __kernel_time_t msg_rtime;  /* last msgrcv time */
    __kernel_time_t msg_ctime;  /* last change time */
    unsigned long msg_lcbytes;  /* Reuse junk fields for 32 bit */
    unsigned long msg_lqbytes;  /* ditto  == 同上... */
    unsigned short msg_cbytes;  /* current number of butes on queue */
    unsigned short msg_qnum;    /* number of messages in queue */
    __kernel_ipc_pid_t msg_lspid; /* pid of last msgsnd */
    __kernel_ipc_pid_t msg_lrpid; /* last receive pid */
}

• 有關命令：

  • ipcs -q 訊息佇列列表
  • ipcrm -q msqid(要刪除的訊息佇列ID)
    示例：
    
    

訊息佇列相關函式。

1、ftok函式

#include <sys/ipc.h>
#include <sys/types.h>
key_t ftok(const char* path, int id);

• ftok 函式把一個已存在的路徑名和一個整數標識轉換成一個key_t值，即IPC關鍵字
• path 引數就是你指定的檔名（已經存在的檔名），一般使用當前目錄。當產生鍵時，只使用id引數的低8位。
• id 是子序號， 只使用8bit （1-255）
• 返回值：若成功返回鍵值，若出錯返回（key_t）-1
  在一般的UNIX實現中，是將檔案的索引節點號取出（inode），前面加上子序號的到key_t的返回值

2、msgget函式

#include <sys/msg.h>
#include <sys/ipc.h>
int msgget(key_t key, int msgflag);

• msgget 通常是呼叫的第一個函式，功能是建立一個新的或已經存在的訊息佇列。此訊息佇列與key相對應。
• key 引數 即ftok函式生成的關鍵字
• flag引數 ：
  
  • IPC_CREAT： 如果IPC不存在，則建立一個IPC資源，否則開啟已存在的IPC。
  • IPC_EXCL ：只有在共享記憶體不存在的時候，新的共享記憶體才建立，否則就產生錯誤。
  • IPC_EXCL與IPC_CREAT一起使用，表示要建立的訊息佇列已經存在。如果該IPC資源存在，則返回-1。
• IPC_EXCL標識本身沒有太大的意義，但是和IPC_CREAT標誌一起使用可以用來保證所得的物件時新建的，而不是開啟已有的物件。
• 返回值 若成功返回訊息佇列ID，若出錯則返回-1

3、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 msgflag);
ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int msgflag);

• msgsnd 將資料放到訊息佇列中 msgrcv 從訊息佇列中讀取資料
• msqid：訊息佇列的識別碼
• msgp：指向訊息緩衝區的指標，用來暫時儲存傳送和接受的訊息。是一個允許使用者定義的通用結構，如下：

struct msgubf
{
    long mtype; // 訊息型別， 必須大於零
    char mtext[SIZE]; // 訊息文字
}

• msgsz：訊息的大小
• ptr 指向一個長整形數，將返回的訊息型別儲存在其中（即結構體 struct msgbuf 的mtype成員）
• nbyte 是存放實際訊息資料的緩衝區的長度
• type ：可以指定想要哪一種訊息
  
  • type == 0 返回佇列的第一個訊息
  • type > 0 返回佇列中訊息型別type的第一個訊息
  • type < 0 返回佇列中訊息型別值小於或等於type絕對值的訊息，如果這種訊息有若干個，則型別值最小的訊息
• msgflag：訊息型別，這個引數是控制函式行為的標識。取值可以是0，標識忽略。
  
  • IPC_NOWAIT，如果訊息佇列為空，則返回一個ENOMSG，並將控制權交回給呼叫函式的程序。如果不指定這個引數，那麼程序將被阻塞知道函式可以從佇列中取得符合條件的訊息為止。
  • 0 表示不關心，忽略此行為
• 返回值：成功執行返回訊息的資料部分的長度，若出錯則返回-1
  msgrcv成功執行時，核心更新與該訊息佇列相關聯的msqid_ds結構以指示呼叫者的程序ID（msg_lrpid）和呼叫時間（msg_rtime），並將佇列中的訊息數（msg_qnum）減1。*

4、msgctl函式

#include <sys/types.g>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid,  int cmd,  struct msqid_ds *buf);

• msgctl 函式 可以直接控制訊息佇列的行為
• msqid 訊息佇列id
• cmd ：命令
  
  • IPC_STAT 讀取訊息佇列的資料結構msqid_ds， 並將其儲存在 buf指定的地址中
  • IPC_SET 設定訊息佇列的資料結構msqid_ds 中的ipc_perm元素的值，這個值取自buf 引數
  • IPC_RMID 從核心中移除訊息佇列。
• 返回值：如果成功返回0，失敗返回-1

程式碼示例

Makefile

client_=client
server_=server

cc=gcc

clientSrc=client.c common.c
serverSrc=server.c common.c

.PHONY:all
all:$(client_) $(server_)

$(client_):$(clientSrc)
    $(cc) -o  [email protected] $^
$(server_):$(serverSrc)
    $(cc) -o  [email protected] $^


.PHONY:clean
clean:
    rm -f $(client_) $(server_) 

common_h

#ifndef _COMMON_H_
#define _COMMON_H_
#include <string.h>  // strcpy
#include <stdio.h>
#include <sys/types.h>
#include <unistd.h>  // read
#include <sys/ipc.h>
#include <sys/msg.h>

#define PATHNAME "./"
#define PROJ_ID  0x666
#define MSGSIZE 1024

#define SERVER_TYPE 1   // 服務端傳送訊息型別
#define CLIENT_TYPE 2   // 客戶端傳送訊息型別

struct msgbuf          // 訊息結構
{
    long mtype;     // 訊息型別
    char mtext[MSGSIZE]; // 訊息buf
};

int createMsgQueue();  // 建立訊息佇列
int destroyMsgQueue( int msqid); // 銷燬訊息佇列

int getMsgQueue();     // 獲取訊息佇列

int sendMsg( int msqid, long type,  const char *_sendInfo);   // 傳送訊息
int recvMsg(int msqid, long type, char buf[]);       // 接收訊息

#endif /* _COMMON_H*/

common_c

#include "common.h"


int commMsg(int msgflag)
{
    // 生成IPC 關鍵字
    key_t _k = ftok(PATHNAME, PROJ_ID);
    int msqid = msgget(_k, msgflag); // 獲取訊息佇列ID
    if(msqid < 0)
    {
        perror("msgget");
        return -2;
    }
    return msqid;

}


int createMsgQueue()  // 建立訊息佇列
{
    return commMsg(IPC_CREAT|IPC_EXCL|0666);
}

int destroyMsgQueue( int msqid) // 銷燬訊息佇列
{
    int _ret = msgctl(msqid, IPC_RMID, 0);
    if(_ret < 0)
    {
        perror("msgctl");
        return -1;
    }
    return 0;
}

int getMsgQueue()     // 獲取訊息佇列
{
    return commMsg(IPC_CREAT);
}

int sendMsg( int msqid, long type,  const char *_sendInfo)         // 傳送訊息
{
    struct msgbuf msg;
    msg.mtype = type;
    strcpy(msg.mtext, _sendInfo);

    int _snd = msgsnd(msqid, &msg, sizeof(msg.mtext), 0);
    if( _snd < 0)
    {
        perror("msgsnd");
        return -1;
    }
    return 0;
}

int recvMsg(int msqid, long type, char buf[])          // 接收訊息
{
    struct msgbuf msg;
    int _rcv = msgrcv(msqid, &msg, sizeof(msg.mtext), type, 0);
    if( _rcv < 0)
    {
        perror("msgrcv");
        return -1;

    }
    strcpy(buf, msg.mtext);
    return 0;
}

client_c

#include "common.h"

void client()
{
    int msqid = getMsgQueue();
    char buf[MSGSIZE];
    while(1)
    {
        printf("Please enter :");
        fflush(stdout);
        ssize_t _s = read(0, buf, sizeof(buf)-1);
        if(_s > 0)
        {
            buf[_s -1] = '\0';
            sendMsg(msqid, CLIENT_TYPE, buf);
        }
        recvMsg(msqid, SERVER_TYPE, buf);
        if(strcmp("exit",buf) == 0)
        {
            printf("服務端退出，客戶端自動退出\n");
            break;
        }
        printf("服務端說：%s\n", buf);
    }
}

int main()
{
    client();
    return 0;
}

server_c

#include "common.h"

void server()
{
    int msqid = createMsgQueue();
    char buf[MSGSIZE];
    while(1)
    {
        // 服務端先接收
        recvMsg(msqid, CLIENT_TYPE, buf); printf("客戶端說：%s\n ", buf);
        printf("Please enter :");
        fflush(stdout);
        ssize_t _s = read(0, buf, sizeof(buf)-1);
        if(_s > 0)
        {
            buf[_s-1] = '\0';
            sendMsg(msqid, SERVER_TYPE, buf);

            if(strcmp(buf, "exit") == 0)
                break;
        }
    }
    destroyMsgQueue(msqid);
}


int main()
{
    server();
    return 0;
}

通訊截圖示例：

注意：如果在啟動server 後 強制結束掉（ctrl+c）程式，則訊息佇列會一直存在，這時再次執行server會執行失敗，需要使用命令 ipcrm -q xxx xx表示要關閉的msqid。

總結：訊息佇列傳送的是資料塊， 生命週期隨核心，依賴於系統介面實現。適用於無血緣關係多個程序之間通訊。