深入理解進程間通信之消息隊列

消息隊列是消息的鏈接表，包括Posix消息隊列system V消息隊列。有足夠權限的進程可以向隊列中添加消息，被賦予讀權限的進程則可以讀走隊列中的消息。消息隊列克服了信號承載信息量少，管道只能承載無格式字節流以及緩沖區大小受限等缺點。消息隊列是隨內核持續的。

IPC持續概念

隨進程持續

IPC一直存在到打開IPC對象的最後一個進程關閉該對象為止。如管道和有名管道；

隨內核持續

IPC一直持續到內核重新自舉或者顯示刪除該對象為止。如消息隊列、信號燈以及共享內存等；

隨文件系統持續

IPC一直持續到顯示刪除該對象為止。

消息隊列的基本概念

　　系統V消息隊列是隨內核持續的，只有在內核重啟或者顯示刪除一個消息隊列時，該消息隊列才會真正被刪除。因此系統中記錄消息隊列的數據結構（struct ipc_ids msg_ids）位於內核中，系統中的所有消息隊列都可以在結構msg_ids中找到訪問入口。

　　消息隊列就是一個消息的鏈表。每個消息隊列都有一個隊列頭，用結構struct msg_queue來描述。隊列頭中包含了該消息隊列的大量信息，包括消息隊列鍵值、用戶ID、組ID、消息隊列中消息數目等等，甚至記錄了最近對消息隊列讀寫進程的ID。讀者可以訪問這些信息，也可以設置其中的某些信息。

結構msg_queue用來描述消息隊列頭，存在於系統空間：

 struct msg_queue {
    struct kern_ipc_perm q_perm;
    time_t q_stime;        /* last msgsnd time */
    time_t q_rtime;        /* last msgrcv time */
    time_t q_ctime;        /* last change time */
    unsigned long q_cbytes;    /* current number of bytes on queue */
    unsigned long q_qnum;      /* number of messages in queue */
    unsigned long q_qbytes;    /* max number of bytes on queue */
    pid_t q_lspid;          /* pid of last msgsnd */
    pid_t q_lrpid;          /* last receive pid */
    struct list_head q_messages;
    struct list_head q_receivers;
    struct list_head q_senders;
};

結構msqid_ds用來設置或返回消息隊列的信息，存在於用戶空間：

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 bytes on queue */
    unsigned short msg_qnum;    /* number of messages in queue */
    unsigned short msg_qbytes;  /* max number of bytes on queue */
    __kernel_ipc_pid_t msg_lspid;  /* pid of last msgsnd */
    __kernel_ipc_pid_t msg_lrpid;  /* last receive pid */
};

下圖說明了內核與消息隊列是怎樣建立起聯系的：

其中：struct ipc_ids msg_ids是內核中記錄消息隊列的全局數據結構；struct msg_queue是每個消息隊列的隊列頭。

從上圖可以看出，全局數據結構 struct ipc_ids msg_ids 可以訪問到每個消息隊列頭的第一個成員：struct kern_ipc_perm；而每個struct kern_ipc_perm能夠與具體的消息隊列對應起來是因為在該結構中，有一個key_t類型成員key，而key則唯一確定一個消息隊列。 kern_ipc_perm結構如下：

struct kern_ipc_perm{  //內核中記錄消息隊列的全局數據結構msg_ids能夠訪問到該結構；
key_t  key;    //該鍵值則唯一對應一個消息隊列
uid_t  uid;
gid_t  gid;
uid_t  cuid;
gid_t  cgid;
mode_t  mode;
unsigned long seq;
}

操作消息隊列

消息隊列操作類型

1、 打開或創建消息隊列

　　消息隊列的內核持續性要求每個消息隊列都在系統范圍內對應唯一的鍵值，所以，要獲得一個消息隊列的描述字，只需提供該消息隊列的鍵值即可；

　　注：消息隊列描述字是由在系統范圍內唯一的鍵值生成的，而鍵值可以看作對應系統內的一條路經。

2、 讀寫操作

　　消息讀寫操作非常簡單，對開發人員來說，每個消息都類似如下的數據結構：

struct msgbuf{
long mtype;
char mtext[1];
};

　　mtype成員代表消息類型，從消息隊列中讀取消息的一個重要依據就是消息的類型；mtext是消息內容，當然長度不一定為1。因此，對於發送消息來說， 首先預置一個msgbuf緩沖區並寫入消息類型和內容，調用相應的發送函數即可；對讀取消息來說，首先分配這樣一個msgbuf緩沖區，然後把消息讀入該緩沖區即可。

3、 獲得或設置消息隊列屬性：

　　消息隊列的信息基本上都保存在消息隊列頭中，因此，可以分配一個類似於消息隊列頭的結構，來返回消息隊列的屬性；同樣可以設置該數據結構。

 

 

消息對列API

1.文件名到鍵值

#include <sys/types.h>
#include <sys/ipc.h>
key_t ftok (char*pathname, char proj)；

它返回與路徑pathname相對應的一個鍵值。該函數不直接對消息隊列操作，但在調用ipc(MSGGET,…)或msgget()來獲得消息隊列描述字前，往往要調用該函數。典型的調用代碼是：

key=ftok(path_ptr, 'a');
    ipc_id=ipc(MSGGET, (int)key, flags,0,NULL,0);
    …

 

2. Linux為操作系統V進程間通信的三種方式（消息隊列、信號燈、共享內存區）提供了一個統一的用戶界面：

int ipc(unsigned int call, int first, int second, int third, void * ptr, long fifth);

第一個參數指明對IPC對象的操作方式，對消息隊列而言共有四種操作：MSGSND、MSGRCV、MSGGET以及MSGCTL，分別代表向消息隊列發送消息、從消息隊列讀取消息、打開或創建消息隊列、控制消息隊列；first參數代表唯一的IPC對象；下面將介紹四種操作。

• int ipc( MSGGET, int first, int second, int third, void* ptr, long fifth);
  與該操作對應的系統V調用為：int msgget( (key_t)first，second)。
• int ipc( MSGCTL, int first, int second, int third, void* ptr, long fifth)
  與該操作對應的系統V調用為：int msgctl( first，second, (struct msqid_ds*) ptr)。
• int ipc( MSGSND, int first, int second, int third, void* ptr, long fifth);
  與該操作對應的系統V調用為：int msgsnd( first, (struct msgbuf*)ptr, second, third)。
• int ipc( MSGRCV, int first, int second, int third, void*ptr, long fifth);
  與該操作對應的系統V調用為：int msgrcv( first，(struct msgbuf*)ptr, second, fifth,third)，

注：本人不主張采用系統調用ipc()，而更傾向於采用系統V或者POSIX進程間通信API。原因如下：

• 雖然該系統調用提供了統一的用戶界面，但正是由於這個特性，它的參數幾乎不能給出特定的實際意義（如以first、second來命名參數），在一定程度上造成開發不便。
• 正如ipc手冊所說的：ipc()是Linux所特有的，編寫程序時應注意程序的移植性問題；
• 該系統調用的實現不過是把系統V IPC函數進行了封裝，沒有任何效率上的優勢；
• 系統V在IPC方面的API數量不多，形式也較簡潔。

3.系統V消息隊列API

　　系統V消息隊列API共有四個，使用時需要包括幾個頭文件：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

1）int msgget(key_t key, int msgflg)

參數key是一個鍵值，由ftok獲得；msgflg參數是一些標志位。該調用返回與健值key相對應的消息隊列描述字。

在以下兩種情況下，該調用將創建一個新的消息隊列：

• 如果沒有消息隊列與健值key相對應，並且msgflg中包含了IPC_CREAT標志位；
• key參數為IPC_PRIVATE；

參數msgflg可以為以下：IPC_CREAT、IPC_EXCL、IPC_NOWAIT或三者的或結果。

調用返回：成功返回消息隊列描述字，否則返回-1。

注：參數key設置成常數IPC_PRIVATE並不意味著其他進程不能訪問該消息隊列，只意味著即將創建新的消息隊列。

2）int msgrcv(int msgid, struct msgbuf *msgp, int msgsz, long msgtyp, int msgflg);

該系統調用從msgid代表的消息隊列中讀取一個消息，並把消息存儲在msgp指向的msgbuf結構中。

msqid為消息隊列描述字；消息返回後存儲在msgp指向的地址，msgsz指定msgbuf的mtext成員的長度（即消息內容的長度），msgtyp為請求讀取的消息類型；讀消息標志msgflg可以為以下幾個常值的或：

• IPC_NOWAIT 如果沒有滿足條件的消息，調用立即返回，此時，errno=ENOMSG
• IPC_EXCEPT 與msgtyp>0配合使用，返回隊列中第一個類型不為msgtyp的消息
• IPC_NOERROR 如果隊列中滿足條件的消息內容大於所請求的msgsz字節，則把該消息截斷，截斷部分將丟失。

msgrcv手冊中詳細給出了消息類型取不同值時(>0; <0; =0)，調用將返回消息隊列中的哪個消息。

msgrcv()解除阻塞的條件有三個：

1. 消息隊列中有了滿足條件的消息；
2. msqid代表的消息隊列被刪除；
3. 調用msgrcv()的進程被信號中斷；

調用返回：成功返回讀出消息的實際字節數，否則返回-1。

3）int msgsnd(int msqid, struct msgbuf *msgp, int msgsz, int msgflg);

向msgid代表的消息隊列發送一個消息，即將發送的消息存儲在msgp指向的msgbuf結構中，消息的大小由msgze指定。

對發送消息來說，有意義的msgflg標志為IPC_NOWAIT，指明在消息隊列沒有足夠空間容納要發送的消息時，msgsnd是否等待。造成msgsnd()等待的條件有兩種：

• 當前消息的大小與當前消息隊列中的字節數之和超過了消息隊列的總容量；
• 當前消息隊列的消息數（單位"個"）不小於消息隊列的總容量（單位"字節數"），此時，雖然消息隊列中的消息數目很多，但基本上都只有一個字節。

msgsnd()解除阻塞的條件有三個：

1. 不滿足上述兩個條件，即消息隊列中有容納該消息的空間；
2. msqid代表的消息隊列被刪除；
3. 調用msgsnd（）的進程被信號中斷；

調用返回：成功返回0，否則返回-1。

4）int msgctl(int msqid, int cmd, struct msqid_ds *buf);

該系統調用對由msqid標識的消息隊列執行cmd操作，共有三種cmd操作：IPC_STAT、IPC_SET 、IPC_RMID。

1. IPC_STAT：該命令用來獲取消息隊列信息，返回的信息存貯在buf指向的msqid結構中；
2. IPC_SET：該命令用來設置消息隊列的屬性，要設置的屬性存儲在buf指向的msqid結構中；可設置屬性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes，同時，也影響msg_ctime成員。
3. IPC_RMID：刪除msqid標識的消息隊列；

調用返回：成功返回0，否則返回-1。

消息隊列的應用實例

#include <sys/types.h>
#include <sys/msg.h>
#include <unistd.h>
void msg_stat(int,struct msqid_ds );

void main()
{
int gflags,sflags,rflags;
key_t key;
int msgid;
int reval;
struct msgsbuf{
        int mtype;
        char mtext[1];
    }msg_sbuf;
struct msgmbuf
    {
            int mtype;
            char mtext[10];
    }msg_rbuf;

struct msqid_ds msg_ginfo,msg_sinfo;
char* msgpath="/unix/msgqueue";
key=ftok(msgpath,'a');
gflags=IPC_CREAT|IPC_EXCL;
msgid=msgget(key,gflags|00666);
if(msgid==-1)
{
            printf("msg create error\n");
            return;
}
//創建一個消息隊列後，輸出消息隊列缺省屬性
msg_stat(msgid,msg_ginfo);
sflags=IPC_NOWAIT;
msg_sbuf.mtype=10;
msg_sbuf.mtext[0]='a';
reval=msgsnd(msgid,&msg_sbuf,sizeof(msg_sbuf.mtext),sflags);
if(reval==-1)
{
            printf("message send error\n");
}
//發送一個消息後，輸出消息隊列屬性
msg_stat(msgid,msg_ginfo);
rflags=IPC_NOWAIT|MSG_NOERROR;
reval=msgrcv(msgid,&msg_rbuf,4,10,rflags);
if(reval==-1)
            printf("read msg error\n");
else
            printf("read from msg queue %d bytes\n",reval);
//從消息隊列中讀出消息後，輸出消息隊列屬性
msg_stat(msgid,msg_ginfo);
msg_sinfo.msg_perm.uid=8;//just a try
msg_sinfo.msg_perm.gid=8;//
msg_sinfo.msg_qbytes=16388;
//此處驗證超級用戶可以更改消息隊列的缺省msg_qbytes
//注意這裡設置的值大於缺省值
reval=msgctl(msgid,IPC_SET,&msg_sinfo);
if(reval==-1)
{
            printf("msg set info error\n");
            return;
}
msg_stat(msgid,msg_ginfo);
//驗證設置消息隊列屬性
reval=msgctl(msgid,IPC_RMID,NULL);//刪除消息隊列
if(reval==-1)
{
            printf("unlink msg queue error\n");
            return;
}
}
void msg_stat(int msgid,struct msqid_ds msg_info)
{
int reval;
sleep(1);//只是為了後面輸出時間的方便
reval=msgctl(msgid,IPC_STAT,&msg_info);
if(reval==-1)
{
            printf("get msg info error\n");
            return;
}
printf("\n");
printf("current number of bytes on queue is %d\n",msg_info.msg_cbytes);
printf("number of messages in queue is %d\n",msg_info.msg_qnum);
printf("max number of bytes on queue is %d\n",msg_info.msg_qbytes);
//每個消息隊列的容量（字節數）都有限制MSGMNB，值的大小因系統而異。在創建新的消息隊列時，
//msg_qbytes的缺省值就是MSGMNB
printf("pid of last msgsnd is %d\n",msg_info.msg_lspid);
printf("pid of last msgrcv is %d\n",msg_info.msg_lrpid);
printf("last msgsnd time is %s", ctime(&(msg_info.msg_stime)));
printf("last msgrcv time is %s", ctime(&(msg_info.msg_rtime)));
printf("last change time is %s", ctime(&(msg_info.msg_ctime)));
printf("msg uid is %d\n",msg_info.msg_perm.uid);
printf("msg gid is %d\n",msg_info.msg_perm.gid);
}