Linux環(huán)境進程間通信（三）
消息隊列
|
|
級別: 初級
鄭彥興
(mlinux@163.com)國防科大計算機學院
2003 年 1 月 17 日
本系列文章中的前兩部分，我們探討管道及信號兩種通信機制，本文將深入第三部分，介紹系統(tǒng) V 消息隊列及其相應 API。
消息隊列（也叫做報文隊列）能夠克服早期unix通信機制的一些缺點。作為早期unix通信機制之一的信號能夠傳送的信息量有限，后來雖然POSIX 1003.1b在信號的實時性方面作了拓廣，使得信號在傳遞信息量方面有了相當程度的改進，但是信號這種通信方式更像"即時"的通信方式，它要求接受信號的進程在某個時間范圍內對信號做出反應，因此該信號最多在接受信號進程的生命周期內才有意義，信號所傳遞的信息是接近于隨進程持續(xù)的概念（process-persistent），見 附錄 1；管道及有名管道及有名管道則是典型的隨進程持續(xù)IPC，并且，只能傳送無格式的字節(jié)流無疑會給應用程序開發(fā)帶來不便，另外，它的緩沖區(qū)大小也受到限制。
消息隊列就是一個消息的鏈表。可以把消息看作一個記錄，具有特定的格式以及特定的優(yōu)先級。對消息隊列有寫權限的進程可以向中按照一定的規(guī)則添加新消息；對消息隊列有讀權限的進程則可以從消息隊列中讀走消息。消息隊列是隨內核持續(xù)的（參見 附錄 1）。
目前主要有兩種類型的消息隊列：POSIX消息隊列以及系統(tǒng)V消息隊列，系統(tǒng)V消息隊列目前被大量使用。考慮到程序的可移植性，新開發(fā)的應用程序應盡量使用POSIX消息隊列。
在本系列專題的序（深刻理解Linux進程間通信（IPC））中，提到對于消息隊列、信號燈、以及共享內存區(qū)來說，有兩個實現(xiàn)版本：POSIX的以及系統(tǒng)V的。Linux內核（內核2.4.18）支持POSIX信號燈、POSIX共享內存區(qū)以及POSIX消息隊列，但對于主流Linux發(fā)行版本之一redhad8.0（內核2.4.18），還沒有提供對POSIX進程間通信API的支持，不過應該只是時間上的事。
因此，本文將主要介紹系統(tǒng)V消息隊列及其相應API。 在沒有聲明的情況下，以下討論中指的都是系統(tǒng)V消息隊列。
一、消息隊列基本概念
- 系統(tǒng)V消息隊列是隨內核持續(xù)的，只有在內核重起或者顯示刪除一個消息隊列時，該消息隊列才會真正被刪除。因此系統(tǒng)中記錄消息隊列的數(shù)據(jù)結構（struct ipc_ids msg_ids）位于內核中，系統(tǒng)中的所有消息隊列都可以在結構msg_ids中找到訪問入口。
- 消息隊列就是一個消息的鏈表。每個消息隊列都有一個隊列頭，用結構struct msg_queue來描述（參見 附錄 2）。隊列頭中包含了該消息隊列的大量信息，包括消息隊列鍵值、用戶ID、組ID、消息隊列中消息數(shù)目等等，甚至記錄了最近對消息隊列讀寫進程的ID。讀者可以訪問這些信息，也可以設置其中的某些信息。
- 下圖說明了內核與消息隊列是怎樣建立起聯(lián)系的：
其中：struct ipc_ids msg_ids是內核中記錄消息隊列的全局數(shù)據(jù)結構；struct msg_queue是每個消息隊列的隊列頭。
從上圖可以看出，全局數(shù)據(jù)結構 struct ipc_ids msg_ids 可以訪問到每個消息隊列頭的第一個成員：struct kern_ipc_perm；而每個struct kern_ipc_perm能夠與具體的消息隊列對應起來是因為在該結構中，有一個key_t類型成員key，而key則唯一確定一個消息隊列。kern_ipc_perm結構如下：
struct kern_ipc_perm{ //內核中記錄消息隊列的全局數(shù)據(jù)結構msg_ids能夠訪問到該結構；
key_t key; //該鍵值則唯一對應一個消息隊列
uid_t uid;
gid_t gid;
uid_t cuid;
gid_t cgid;
mode_t mode;
unsigned long seq;
}
|
二、操作消息隊列
對消息隊列的操作無非有下面三種類型：
1、 打開或創(chuàng)建消息隊列
消息隊列的內核持續(xù)性要求每個消息隊列都在系統(tǒng)范圍內對應唯一的鍵值，所以，要獲得一個消息隊列的描述字，只需提供該消息隊列的鍵值即可；
注：消息隊列描述字是由在系統(tǒng)范圍內唯一的鍵值生成的，而鍵值可以看作對應系統(tǒng)內的一條路經(jīng)。
2、 讀寫操作
消息讀寫操作非常簡單，對開發(fā)人員來說，每個消息都類似如下的數(shù)據(jù)結構：
struct msgbuf{
long mtype;
char mtext[1];
};
|
mtype成員代表消息類型，從消息隊列中讀取消息的一個重要依據(jù)就是消息的類型；mtext是消息內容，當然長度不一定為1。因此，對于發(fā)送消息來說，首先預置一個msgbuf緩沖區(qū)并寫入消息類型和內容，調用相應的發(fā)送函數(shù)即可；對讀取消息來說，首先分配這樣一個msgbuf緩沖區(qū)，然后把消息讀入該緩沖區(qū)即可。
3、 獲得或設置消息隊列屬性：
消息隊列的信息基本上都保存在消息隊列頭中，因此，可以分配一個類似于消息隊列頭的結構(struct msqid_ds，見 附錄 2)，來返回消息隊列的屬性；同樣可以設置該數(shù)據(jù)結構。
消息隊列API
1、文件名到鍵值
#include <sys/types.h>
#include <sys/ipc.h>
key_t ftok (char*pathname, char proj)；
|
它返回與路徑pathname相對應的一個鍵值。該函數(shù)不直接對消息隊列操作，但在調用ipc(MSGGET,…)或msgget()來獲得消息隊列描述字前，往往要調用該函數(shù)。典型的調用代碼是：
key=ftok(path_ptr, 'a');
ipc_id=ipc(MSGGET, (int)key, flags,0,NULL,0);
…
|
2、linux為操作系統(tǒng)V進程間通信的三種方式（消息隊列、信號燈、共享內存區(qū)）提供了一個統(tǒng)一的用戶界面：
int ipc
(unsigned int call, int first, int second, int third, void * ptr, long fifth);
第一個參數(shù)指明對IPC對象的操作方式，對消息隊列而言共有四種操作：MSGSND、MSGRCV、MSGGET以及MSGCTL，分別代表向消息隊列發(fā)送消息、從消息隊列讀取消息、打開或創(chuàng)建消息隊列、控制消息隊列；first參數(shù)代表唯一的IPC對象；下面將介紹四種操作。
-
int ipc( MSGGET, intfirst, intsecond, intthird, void*ptr, longfifth);
與該操作對應的系統(tǒng)V調用為：int msgget( (key_t)first，second)。
-
int ipc( MSGCTL, intfirst, intsecond, intthird, void*ptr, longfifth)
與該操作對應的系統(tǒng)V調用為：int msgctl( first，second, (struct msqid_ds*) ptr)。
-
int ipc( MSGSND, intfirst, intsecond, intthird, void*ptr, longfifth);
與該操作對應的系統(tǒng)V調用為：int msgsnd( first, (struct msgbuf*)ptr, second, third)。
-
int ipc( MSGRCV, intfirst, intsecond, intthird, void*ptr, longfifth);
與該操作對應的系統(tǒng)V調用為：int msgrcv( first，(struct msgbuf*)ptr, second, fifth,third)，
注：本人不主張采用系統(tǒng)調用ipc()，而更傾向于采用系統(tǒng)V或者POSIX進程間通信API。原因如下：
- 雖然該系統(tǒng)調用提供了統(tǒng)一的用戶界面，但正是由于這個特性，它的參數(shù)幾乎不能給出特定的實際意義（如以first、second來命名參數(shù)），在一定程度上造成開發(fā)不便。
- 正如ipc手冊所說的：ipc()是linux所特有的，編寫程序時應注意程序的移植性問題；
- 該系統(tǒng)調用的實現(xiàn)不過是把系統(tǒng)V IPC函數(shù)進行了封裝，沒有任何效率上的優(yōu)勢；
- 系統(tǒng)V在IPC方面的API數(shù)量不多，形式也較簡潔。
3.系統(tǒng)V消息隊列API
系統(tǒng)V消息隊列API共有四個，使用時需要包括幾個頭文件：
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
|
1）int msgget(key_t key, int msgflg)
參數(shù)key是一個鍵值，由ftok獲得；msgflg參數(shù)是一些標志位。該調用返回與健值key相對應的消息隊列描述字。
在以下兩種情況下，該調用將創(chuàng)建一個新的消息隊列：
- 如果沒有消息隊列與健值key相對應，并且msgflg中包含了IPC_CREAT標志位；
- key參數(shù)為IPC_PRIVATE；
參數(shù)msgflg可以為以下：IPC_CREAT、IPC_EXCL、IPC_NOWAIT或三者的或結果。
調用返回：成功返回消息隊列描述字，否則返回-1。
注：參數(shù)key設置成常數(shù)IPC_PRIVATE并不意味著其他進程不能訪問該消息隊列，只意味著即將創(chuàng)建新的消息隊列。
2）int msgrcv(int msqid, struct msgbuf *msgp, int msgsz, long msgtyp, int msgflg);
該系統(tǒng)調用從msgid代表的消息隊列中讀取一個消息，并把消息存儲在msgp指向的msgbuf結構中。
msqid為消息隊列描述字；消息返回后存儲在msgp指向的地址，msgsz指定msgbuf的mtext成員的長度（即消息內容的長度），msgtyp為請求讀取的消息類型；讀消息標志msgflg可以為以下幾個常值的或：
- IPC_NOWAIT 如果沒有滿足條件的消息，調用立即返回，此時，errno=ENOMSG
- IPC_EXCEPT 與msgtyp>0配合使用，返回隊列中第一個類型不為msgtyp的消息
- IPC_NOERROR 如果隊列中滿足條件的消息內容大于所請求的msgsz字節(jié)，則把該消息截斷，截斷部分將丟失。
msgrcv手冊中詳細給出了消息類型取不同值時(>0; <0; =0)，調用將返回消息隊列中的哪個消息。
msgrcv()解除阻塞的條件有三個：
- 消息隊列中有了滿足條件的消息；
- msqid代表的消息隊列被刪除；
- 調用msgrcv（）的進程被信號中斷；
調用返回：成功返回讀出消息的實際字節(jié)數(shù)，否則返回-1。
3）int msgsnd(int msqid, struct msgbuf *msgp, int msgsz, int msgflg);
向msgid代表的消息隊列發(fā)送一個消息，即將發(fā)送的消息存儲在msgp指向的msgbuf結構中，消息的大小由msgze指定。
對發(fā)送消息來說，有意義的msgflg標志為IPC_NOWAIT，指明在消息隊列沒有足夠空間容納要發(fā)送的消息時，msgsnd是否等待。造成msgsnd()等待的條件有兩種：
- 當前消息的大小與當前消息隊列中的字節(jié)數(shù)之和超過了消息隊列的總容量；
- 當前消息隊列的消息數(shù)（單位"個"）不小于消息隊列的總容量（單位"字節(jié)數(shù)"），此時，雖然消息隊列中的消息數(shù)目很多，但基本上都只有一個字節(jié)。
msgsnd()解除阻塞的條件有三個：
- 不滿足上述兩個條件，即消息隊列中有容納該消息的空間；
- msqid代表的消息隊列被刪除；
- 調用msgsnd（）的進程被信號中斷；
調用返回：成功返回0，否則返回-1。
4）int msgctl(int msqid, int cmd, struct msqid_ds *buf);
該系統(tǒng)調用對由msqid標識的消息隊列執(zhí)行cmd操作，共有三種cmd操作：IPC_STAT、IPC_SET 、IPC_RMID。
- IPC_STAT：該命令用來獲取消息隊列信息，返回的信息存貯在buf指向的msqid結構中；
- IPC_SET：該命令用來設置消息隊列的屬性，要設置的屬性存儲在buf指向的msqid結構中；可設置屬性包括：msg_perm.uid、msg_perm.gid、msg_perm.mode以及msg_qbytes，同時，也影響msg_ctime成員。
- IPC_RMID：刪除msqid標識的消息隊列；
調用返回：成功返回0，否則返回-1。
三、消息隊列的限制
每個消息隊列的容量（所能容納的字節(jié)數(shù)）都有限制，該值因系統(tǒng)不同而不同。在后面的應用實例中，輸出了redhat 8.0的限制，結果參見 附錄 3。
另一個限制是每個消息隊列所能容納的最大消息數(shù)：在redhad 8.0中，該限制是受消息隊列容量制約的：消息個數(shù)要小于消息隊列的容量（字節(jié)數(shù)）。
注：上述兩個限制是針對每個消息隊列而言的，系統(tǒng)對消息隊列的限制還有系統(tǒng)范圍內的最大消息隊列個數(shù)，以及整個系統(tǒng)范圍內的最大消息數(shù)。一般來說，實際開發(fā)過程中不會超過這個限制。
四、消息隊列應用實例
消息隊列應用相對較簡單，下面實例基本上覆蓋了對消息隊列的所有操作，同時，程序輸出結果有助于加深對前面所講的某些規(guī)則及消息隊列限制的理解。
#include <sys/types.h>
#include <sys/msg.h>
#include <unistd.h>
void msg_stat(int,struct msqid_ds );
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;
}
//創(chuàng)建一個消息隊列后，輸出消息隊列缺省屬性
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");
}
//發(fā)送一個消息后，輸出消息隊列屬性
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);
//每個消息隊列的容量（字節(jié)數(shù)）都有限制MSGMNB，值的大小因系統(tǒng)而異。在創(chuàng)建新的消息隊列時，//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);
}
|
程序輸出結果見
附錄 3。
小結：
消息隊列與管道以及有名管道相比，具有更大的靈活性，首先，它提供有格式字節(jié)流，有利于減少開發(fā)人員的工作量；其次，消息具有類型，在實際應用中，可作為優(yōu)先級使用。這兩點是管道以及有名管道所不能比的。同樣，消息隊列可以在幾個進程間復用，而不管這幾個進程是否具有親緣關系，這一點與有名管道很相似；但消息隊列是隨內核持續(xù)的，與有名管道（隨進程持續(xù)）相比，生命力更強，應用空間更大。
附錄 1： 在參考文獻[1]中，給出了IPC隨進程持續(xù)、隨內核持續(xù)以及隨文件系統(tǒng)持續(xù)的定義：
- 隨進程持續(xù)：IPC一直存在到打開IPC對象的最后一個進程關閉該對象為止。如管道和有名管道；
- 隨內核持續(xù)：IPC一直持續(xù)到內核重新自舉或者顯示刪除該對象為止。如消息隊列、信號燈以及共享內存等；
- 隨文件系統(tǒng)持續(xù)：IPC一直持續(xù)到顯示刪除該對象為止。
附錄 2：
結構msg_queue用來描述消息隊列頭，存在于系統(tǒng)空間：
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 */
};
|
//可以看出上述兩個結構很相似。
附錄 3： 消息隊列實例輸出結果：
current number of bytes on queue is 0
number of messages in queue is 0
max number of bytes on queue is 16384
pid of last msgsnd is 0
pid of last msgrcv is 0
last msgsnd time is Thu Jan 1 08:00:00 1970
last msgrcv time is Thu Jan 1 08:00:00 1970
last change time is Sun Dec 29 18:28:20 2002
msg uid is 0
msg gid is 0
//上面剛剛創(chuàng)建一個新消息隊列時的輸出
current number of bytes on queue is 1
number of messages in queue is 1
max number of bytes on queue is 16384
pid of last msgsnd is 2510
pid of last msgrcv is 0
last msgsnd time is Sun Dec 29 18:28:21 2002
last msgrcv time is Thu Jan 1 08:00:00 1970
last change time is Sun Dec 29 18:28:20 2002
msg uid is 0
msg gid is 0
read from msg queue 1 bytes
//實際讀出的字節(jié)數(shù)
current number of bytes on queue is 0
number of messages in queue is 0
max number of bytes on queue is 16384 //每個消息隊列最大容量（字節(jié)數(shù)）
pid of last msgsnd is 2510
pid of last msgrcv is 2510
last msgsnd time is Sun Dec 29 18:28:21 2002
last msgrcv time is Sun Dec 29 18:28:22 2002
last change time is Sun Dec 29 18:28:20 2002
msg uid is 0
msg gid is 0
current number of bytes on queue is 0
number of messages in queue is 0
max number of bytes on queue is 16388 //可看出超級用戶可修改消息隊列最大容量
pid of last msgsnd is 2510
pid of last msgrcv is 2510 //對操作消息隊列進程的跟蹤
last msgsnd time is Sun Dec 29 18:28:21 2002
last msgrcv time is Sun Dec 29 18:28:22 2002
last change time is Sun Dec 29 18:28:23 2002 //msgctl()調用對msg_ctime有影響
msg uid is 8
msg gid is 8
|
參考資料
- UNIX網(wǎng)絡編程第二卷：進程間通信，作者：W.Richard Stevens，譯者：楊繼張，清華大學出版社。對POSIX以及系統(tǒng)V消息隊列都有闡述，對Linux環(huán)境下的程序開發(fā)有極大的啟發(fā)意義。
- linux內核源代碼情景分析（上），毛德操、胡希明著，浙江大學出版社，給出了系統(tǒng)V消息隊列相關的源代碼分析。
- http://www.fanqiang.com/a4/b2/20010508/113315.html，主要闡述linux下對文件的操作，詳細介紹了對文件的存取權限位，對IPC對象的存取權限同樣具有很好的借鑒意義。
- msgget、msgsnd、msgrcv、msgctl手冊
關于作者
|
| | 鄭彥興，男，現(xiàn)攻讀國防科大計算機學院網(wǎng)絡方向博士學位。您可以通過電子郵件 mlinux@163.com和他聯(lián)系。 |