在Linux中，IPC消息隊列是一個雙向通信的全內(nèi)存設(shè)計，即內(nèi)核保證了讀寫順序和數(shù)據(jù)同步，并且是性能比較好的先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)。消息隊列的應(yīng)用場景：比如異步任務(wù)處理，搶占式的數(shù)據(jù)分發(fā)，順序緩存區(qū)等。

消息隊列的產(chǎn)生原因

消息隊列其實就是消息傳輸過程中保存消息的容器，既然有了管道，為什么要出現(xiàn)消息隊列呢？理由如下：

（1）生命周期：匿名管道和命名管道都是隨進(jìn)程的，意味著管道的生命周期是隨進(jìn)程的退出而退出的。

（2）傳送方式：管道傳送數(shù)據(jù)時以無格式字節(jié)流的形式傳送，給程序的開發(fā)帶來不便。

（3）信號傳遞量：擔(dān)當(dāng)數(shù)據(jù)傳送媒介的管道，其緩沖區(qū)大小受到較大的限制。

鑒于上面的三種原因：IPC方式下的另一種進(jìn)程間通信——>消息隊列應(yīng)運(yùn)而生。它克服了信號傳遞信息少、管道只能承載無格式字節(jié)流以及緩沖區(qū)大小受限等缺點。

什么是消息隊列？

[普通定義]：消息隊列提供了一種從一個進(jìn)程向另一個進(jìn)程發(fā)送一個數(shù)據(jù)塊的方法，每個數(shù)據(jù)塊都被認(rèn)為含有一個類型，接受者接受的數(shù)據(jù)塊可以有不同的類型值，我們可以通過發(fā)送消息來避免命名管道的同步和阻塞問題。

[最佳定義]：內(nèi)核地址空間中的內(nèi)部鏈表，消息可以順序地發(fā)送到隊列中，并以幾種不同的方式從隊列中獲取，每個消息隊列是由IPC標(biāo)識符所唯一標(biāo)識的。

消息隊列與管道的不同在于：消息隊列是基于消息的，而管道是基于字節(jié)流的，且消息隊列的讀取不一定是先入先出，而且如果你沒有顯示的刪除它，那么在關(guān)機(jī)之前它一直存在。

消息隊列的上限

消息隊列與管道也是一樣的不足，就是每個消息隊列的最大長度是有上限的（MSGMAX），每個消息隊列的總的字節(jié)數(shù)是有上限（MSGMNB），系統(tǒng)上消息隊列的總數(shù)也是有一個上限（MSGMNI）。

[root@localhost panpan]# cat /proc/sys/kernel/msgmax 65536 //每個消息的最大長度 [root@localhost panpan]# cat /proc/sys/kernel/msgmnb 65536 //每個消息隊列的總的字節(jié)數(shù) [root@localhost panpan]# cat /proc/sys/kernel/msgmni 1735 //系統(tǒng)消息隊列的總數(shù)

IPC對象數(shù)據(jù)結(jié)構(gòu)

內(nèi)核為每個IPC對象維護(hù)了一個數(shù)據(jù)結(jié)構(gòu)（存在于/usr/include/linux/ipc.h）

struct ipc_perm {__kernel_key_t key;//端口號__kernel_uid_t uid;//所有者的用戶ID__kernel_gid_t gid;//所有者組ID__kernel_uid_t cuid;//創(chuàng)建者的用戶ID__kernel_gid_t cgid;//創(chuàng)建者的組ID__kernel_mode_t mode; //訪問模式unsigned short seq;//順序值 };

不僅如此，消息隊列，共享內(nèi)存和信號量都有這樣一個共同的數(shù)據(jù)結(jié)構(gòu)。

消息隊列結(jié)構(gòu)

消息隊列結(jié)構(gòu)存在于/usr/include/linux/msg.h目錄下：

struct msqid_ds {struct ipc_perm msg_perm; /* IPC對象結(jié)構(gòu)體 */struct msg *msg_first; /*消息隊列頭指針*/struct msg *msg_last; /*消息隊列尾指針*/__kernel_time_t msg_stime; /*最后一次插入消息隊列消息的時間*/__kernel_time_t msg_rtime; /*最后一次接收消息即刪除隊列中一個消息的時間*/__kernel_time_t msg_ctime; /* 最后修改隊列的時間*/unsigned long msg_lcbytes; /* Reuse junk fields for 32 bit */unsigned long msg_lqbytes; /* ditto */unsigned short msg_cbytes; /*隊列上所有消息總的字節(jié)數(shù) */unsigned short msg_qnum; /*在當(dāng)前隊列上消息的個數(shù) */unsigned short msg_qbytes; /* 隊列最大的字節(jié)數(shù) */__kernel_ipc_pid_t msg_lspid;/* 發(fā)送最后一條消息的進(jìn)程的pid */__kernel_ipc_pid_t msg_lrpid;/* 接收最后一條消息的進(jìn)程的pid */ };

從上圖可以看出，消息隊列結(jié)構(gòu)體中的第一條內(nèi)容就是IPC結(jié)構(gòu)體，即IPC結(jié)構(gòu)體是共用的，后面的都是消息隊列所有的成員，并且消息隊列是由鏈表來實現(xiàn)的。

消息隊列相關(guān)函數(shù)接口說明

1.msgget ( )函數(shù)（用于創(chuàng)建和訪問一個消息隊列）

int msgget(key_t key, int msgflg);

參數(shù)[key]：類似于端口號，也可以由ftok函數(shù)生成。

參數(shù)[msgflg]：存在兩個IPC標(biāo)志，即IPC_CREAT和IPC_EXCL。

（1）IPC_CREAT：如果IPC不存在，則創(chuàng)建一個IPC資源，否則打開操作。

（2）IPC_EXCL:只有在共享內(nèi)存不存在的時候，新的共享內(nèi)存才建立，否則出錯。如果單獨(dú)使用IPC_EXCL，xxxget ( )函數(shù)要么返回一個已經(jīng)存在的共享內(nèi)存的操作符，要么返回一個新建的共享內(nèi)存的標(biāo)識符。

（3）如果IPC_CREAT和IPC_EXCL同時使用，xxxget ( ) 函數(shù)將返回一個新建的IPC標(biāo)識符，如果該IPC資源已經(jīng)存在，則返回-1出錯，這樣就保證了只有是二者同時使用，我們就可以保證所得的對象一定是新建的。

2.msgsnd ( ) 和 msgrcv ( ) 函數(shù)（msgrcv從隊列中取出消息，msgsnd將數(shù)據(jù)放到消息隊列中）。

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);

參數(shù)[msgid]：由msgget函數(shù)返回的消息隊列標(biāo)識符。

參數(shù)[msgp]：指向一個準(zhǔn)備發(fā)送消息的指針，此位置用來暫時存儲發(fā)送和接受的消息：是一個用戶可以定義的通用結(jié)構(gòu)體，形態(tài)如下：

struct msgstu {long type; //大于0char mtext[用戶指定大小]; };

參數(shù)[msgsz]：指msgp指向的消息的長度。

參數(shù)[msgtyp]：從消息隊列內(nèi)讀取的消息形態(tài)，如果值為0，則表示消息隊列中的所有消息都會被讀取。

參數(shù)[msgflag]：用來執(zhí)行核心程序在隊列沒有數(shù)據(jù)的情況下采取的行動，如果msgflg和常數(shù)IPC_NOWAIT合用，則在msgsnd ( ) 執(zhí)行時，若消息隊列已滿，則msgsnd ( )就不會被阻塞，而會立即返回-1；如果執(zhí)行的是msgrcv ( ) ，則在消息隊列為空時，不做等待馬上返回-1，并設(shè)定錯誤碼為ENOMSG；當(dāng)msgflg為0時，msgsnd ( ) 以及 msgrcv ( ) 在隊列為滿或者為空的情況下，采取阻塞等待的處理模式。

3.msgctl ( ) 函數(shù)（控制消息隊列，與共享內(nèi)存的shmctl類似）

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

參數(shù)[msgid]：由msgget函數(shù)返回的消息隊列標(biāo)識符。

參數(shù)[cmd]：指將要采取的行動，系統(tǒng)定義了3中cmd操作：

IPC_STAT：該命令用來獲取消息隊列對應(yīng)的msgid_ds數(shù)據(jù)結(jié)構(gòu)，并將保存到buf指定的地址空間。

IPC_SET：設(shè)定消息隊列的屬性，要設(shè)置的屬性存儲在buf里面。

IPC_RMID:從內(nèi)核中刪除msgid表示的消息隊列。

參數(shù)[buf]：指向msg_ds結(jié)構(gòu)的指針，它指向消息隊列模式和訪問權(quán)限的結(jié)構(gòu)。

msg_ds結(jié)構(gòu)體的內(nèi)容如下：

struct msgid_ds { uid_t shm_perm.uid; uid_t shm_perm.gid; mode_t shm_perm.mode; };

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

key_t鍵

System_v IPC使用key_t鍵作為他們的名字，在CentOS6.5 key_t的值被定義為int類型。

/usr/include/sys/ipc.h

typedef __key_t key_t;

/usr/include/bits/types.h

__STD_TYPE __DADDR_T_TYPE __daddr_t; /* The type of a disk address. */ __STD_TYPE __SWBLK_T_TYPE __swblk_t; /* Type of a swap block maybe? */ __STD_TYPE __KEY_T_TYPE __key_t; /* Type of an IPC key. */ /usr/include/bits/typesizes.h

#define __KEY_T_TYPE __S32_TYPE

/usr/include/bits/types

# define __S64_TYPE long int # define __U64_TYPE unsigned long int

ftok ( ) 函數(shù)

函數(shù)ftok把一個已經(jīng)存在的路徑名和一個整數(shù)標(biāo)識符轉(zhuǎn)換成一個key_t值，稱為IPC鍵。

key_t ftok(const char *pathname, int proj_id);

參數(shù)[pathname]：通常是跟本應(yīng)用有關(guān)的目錄。

參數(shù)[proj_id]：指的是本應(yīng)用所用到的IPC的一個序列號，成功返回IPC鍵，失敗返回-1。

注：兩個進(jìn)程如在pathname和proj_id上達(dá)成一致（即約定好），雙方就都能夠通過調(diào)用ftok函數(shù)得到同一個IPC鍵。

pathname的實現(xiàn)是組合了三個鍵，分別是：

（1）pathname所在文件系統(tǒng)的信息（stat結(jié)構(gòu)的st_dev成員）。

（2）pathname在本文件系統(tǒng)內(nèi)的索引節(jié)點號（stat結(jié)構(gòu)體st_ino成員）。

（3）id的低序8位（不能為0）。

ftok調(diào)用返回的整數(shù)IPC鍵由proj_id的低序8位，st_dev成員的低序8位，st_info的低序16位組合而成。

不能保證兩個不同的路徑名與同一個proj_id的組合產(chǎn)生不同的鍵，因為上面所列的三個條目（文件系統(tǒng)標(biāo)識符、索引節(jié)點、proj_id）中的信息位數(shù)可能大于一個整數(shù)的信息位數(shù)。

消息隊列模擬實現(xiàn)客戶端與服務(wù)器端通信

Makefile

//Makefile .PHONY:all all:client serverclient:client.c comm.cgcc -o $@ $^server:server.c comm.cgcc -o $@ $^.PHONY:clean clean:rm -f client server

comm.h

//comm.h #ifndef __COMM_H__ #define __COMM_H__#include<stdio.h> #include<sys/types.h> #include<sys/ipc.h> #include<sys/msg.h> #include<string.h>#define PATHNAME "." #define PROJ_ID 0x6666#define SERVER_TYPE 1 #define CLIENT_TYPE 2struct msgbuf {long mtype;char mtext[1024]; }; int CreateMsgQueue();//創(chuàng)建消息隊列 int GetMsgQueue();//接受消息隊列 int DestroyMsgQueue(int msgid);//銷毀消息隊列 int SendMsg(int msgid,int who,char* msg);//發(fā)送消息 int RecvMsg(int msgid,int recvType,char out[]);//接受消息 #endif

comm.c

//comm.c #include"comm.h"//success > 0 failed == -1static int CommMsgQueue(int flags) {key_t key = ftok(PATHNAME,PROJ_ID);if(key < 0){perror("ftok");return -1;}int msgid = msgget(key,flags);if(msgid < 0){perror("msgget");}return msgid; } int CreateMsgQueue() {return CommMsgQueue(IPC_CREAT | IPC_EXCL | 0666); } int GetMsgQueue() {return CommMsgQueue(IPC_CREAT); } int DestroyMsgQueue(int msgid) {if(msgctl(msgid,IPC_RMID,NULL) < 0){perror("msgctl");return -1;}return 0; } int SendMsg(int msgid,int who,char* msg) {struct msgbuf buf;buf.mtype = who;strcpy(buf.mtext,msg);if(msgsnd(msgid,(void*)&buf,sizeof(buf.mtext),0) < 0){perror("msgsnd");return -1;}return 0; } int RecvMsg(int msgid,int recvType,char out[]) {struct msgbuf buf;if(msgrcv(msgid,(void*)&buf,sizeof(buf.mtext),recvType,0) < 0){perror("msgrcv");return -1;}strcpy(out,buf.mtext);return 0; }

server.c

//server.c #include"comm.h"int main() {int msgid = CreateMsgQueue();char buf[1024];while(1){buf[0] = 0;RecvMsg(msgid,CLIENT_TYPE,buf);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;SendMsg(msgid,SERVER_TYPE,buf);printf("send done,wait recv ...\n");}}DestroyMsgQueue(msgid);return 0; }

client.c

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

顯示IPC資源：

[panpan@localhost msg]$ ipcs------ Shared Memory Segments -------- key shmid owner perms bytes nattch status 0x00000000 131072 panpan 600 393216 2 dest 0x00000000 163841 panpan 600 393216 2 dest 0x00000000 196610 panpan 600 393216 2 dest 0x00000000 229379 panpan 600 393216 2 dest 0x00000000 262148 panpan 600 393216 2 dest 0x00000000 294917 panpan 600 393216 2 dest 0x00000000 327686 panpan 600 393216 2 dest 0x00000000 360455 panpan 600 393216 2 dest 0x00000000 393224 panpan 600 393216 2 dest 0x00000000 425993 panpan 600 393216 2 dest 0x00000000 458762 panpan 600 393216 2 dest 0x00000000 491531 panpan 600 393216 2 dest 0x00000000 524300 panpan 600 393216 2 dest 0x00000000 557069 panpan 600 393216 2 dest 0x00000000 589838 panpan 600 393216 2 dest 0x00000000 622607 panpan 600 393216 2 dest ------ Semaphore Arrays -------- key semid owner perms nsems ------ Message Queues -------- key msqid owner perms used-bytes messages 0x6602248a 32768 panpan 666 0 0

刪除IPC資源：

[panpan@localhost msg]$ ipcrm -q 32768 [panpan@localhost msg]$ ipcs------ Shared Memory Segments -------- key shmid owner perms bytes nattch status 0x00000000 131072 panpan 600 393216 2 dest 0x00000000 163841 panpan 600 393216 2 dest 0x00000000 196610 panpan 600 393216 2 dest 0x00000000 229379 panpan 600 393216 2 dest 0x00000000 262148 panpan 600 393216 2 dest 0x00000000 294917 panpan 600 393216 2 dest 0x00000000 327686 panpan 600 393216 2 dest 0x00000000 360455 panpan 600 393216 2 dest 0x00000000 393224 panpan 600 393216 2 dest 0x00000000 425993 panpan 600 393216 2 dest 0x00000000 458762 panpan 600 393216 2 dest 0x00000000 491531 panpan 600 393216 2 dest 0x00000000 524300 panpan 600 393216 2 dest 0x00000000 557069 panpan 600 393216 2 dest 0x00000000 589838 panpan 600 393216 2 dest 0x00000000 622607 panpan 600 393216 2 dest ------ Semaphore Arrays -------- key semid owner perms nsems ------ Message Queues -------- key msqid owner perms used-bytes messages

運(yùn)行結(jié)果：

總結(jié)：

（1）消息隊列可以獨(dú)立于發(fā)送和接收數(shù)據(jù)的進(jìn)程而存在，從而消除了在同步命名管道的打開和關(guān)閉可能產(chǎn)生的困難。

（2）可以通過發(fā)送消息可以避免命名管道的同步和阻塞問題，不需要由進(jìn)程自己來提供同步方法。

（3）接受程序可以通過消息類型有選擇的接收數(shù)據(jù)，而不像命名管道中的那樣，只能默認(rèn)的接受。

（4）雙向通信，生命周期隨內(nèi)核。

總結(jié)

以上是生活随笔為你收集整理的进程间通信——消息队列（Message queue）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。