POSIX 訊息佇列

1.概述

訊息佇列可認為是一個訊息連結串列，頭訊息指定當前佇列的兩個屬性：佇列中允許的最大訊息數和每個訊息的最大大小。而每個訊息本身除了資料之外還包括優先順序和資料長度等內容。

1.1POSIX訊息佇列和SystemV訊息佇列的區別

1.對於Posix訊息佇列讀操作總是返回最高優先順序的最早訊息，而SystemV則可以返回任意指定優先順序的訊息。
2.當往一個空佇列放置訊息時，Posix訊息佇列允許產生一個訊號或者啟動一個執行緒，SystemV不具備這樣的機制。

1.2Posix訊息佇列資料結構

設想訊息佇列是一個連結串列，連結串列的頭節點包含了最大訊息數mq_maxmsg

2.POSIX訊息佇列程式設計

POSIX 訊息佇列的IPC物件是以檔案的方式存在於虛擬檔案系統中，這些檔案可以在終端使用ls命令檢視，和使用rm命令刪除。當我們要操作這些訊息佇列時，需要掛在訊息佇列檔案系統：

su
Password:
# mkdir /dev/mqueue
# mount -t mqueue none /dev/mqueue
$exit

2.1 mq_open、mq_close和mq_unlink

#include <fcntl.h>           /* For O_* constants */
 

#include <sys/stat.h>        /* For mode constants */
#include <mqueue.h>

mqd_t mq_open(const char *name, int oflag);
mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);

Link with -lrt.

mq_open()建立或者開啟一個已經存在的POSIX 訊息佇列，這些訊息佇列用name引數來標識，一般是”/xxxx”的形式，建立成功後，訊息佇列將存在掛載在檔案系統上的目錄，例如/dev/mqueue

#include <mqueue.h>

int mq_close(mqd_t mqdes);

Link with -lrt.

與close()函式的功能類似，關閉描述符，但是訊息佇列仍然存在，要刪除訊息佇列，需要使用mq_unlink()

#include <mqueue.h>

int mq_unlink(const char *name);

Link with -lrt

2.2 訊息佇列屬性

每個訊息佇列有四個屬性，如下結構體所示：

struct mq_attr {
    long mq_flags;       /* Flags: 0 or O_NONBLOCK */
    long mq_maxmsg;      /* Max number of messages on queue */
    long mq_msgsize;     /* Max size of a message (in bytes) */
    long mq_curmsgs;     /*numbers of messages currently in queue */
};

這些屬性的預設值可以通過下述方式檢視：

cat /proc/sys/fs/mqueue/msg_max         // 對應mq_maxmsg 預設10
cat /proc/sys/fs/mqueue/msgsize_max     //對應mq_msgsize 預設8192
cat /proc/sys/fs/mqueue/queues_max      //最大訊息佇列數 預設256

也可以通過mq_setattr和mq_getattr這兩個函式來設定和獲取狀態：

#include <mqueue.h>

int mq_getattr(mqd_t mqdes, struct mq_attr *attr);

int mq_setattr(mqd_t mqdes, struct mq_attr *newattr, struct mq_attr *oldattr);

Link with -lrt.

2.3 mq_send和mq_receive函式

#include <mqueue.h>
int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int  msg_prio);
ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);

跟write和read類似，不同的是，mq_send在傳送訊息時最後一個引數作為訊息的優先順序，mq_receive最後一個引數指標用來返回訊息優先順序。如果優先順序並不需要，那麼將unsigned int msg_prio設定為0，unsigned int *msg_prio設定為NULL。

2.4完整測試程式碼

簡單的讀寫，寫入端依次寫入優先順序不同的五個訊息，讀入端依次讀出，根據POSIX訊息佇列的定義，讀出順序將是依次讀出優先順序高的最早訊息。

寫入端

//psxmq_send.c

#include <mqueue.h>
#include <stdbool.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>    
#include <time.h>   
#include <errno.h>
#include <string.h>

#define MAXSIZE 1024
int main(int argc,char**argv)
{
    if(argc!=2)
    {
    printf("Usage: %s /mqname \r\n",argv[0]);
    return -1;
    }
    char Msg[MAXSIZE]="Hello World";
    char *name = argv[1];
    int flags = O_RDWR | O_CREAT | O_EXCL ;
    mode_t mode = S_IRUSR | S_IWUSR| S_IRGRP |S_IROTH;
    //set mq_attr
    struct mq_attr attr;
    attr.mq_flags=0;
    attr.mq_maxmsg=10;
    attr.mq_msgsize=sizeof(Msg);
    attr.mq_curmsgs=0;

    mqd_t mqid = mq_open(name,flags,mode,&attr);
    if(mqid==-1)
    {
        printf("error %s (%d)\r\n",strerror(errno),errno);
        return -1;
    }
    int i;
    for(i=0;i<5;i++)
    {
        if(mq_send(mqid,Msg,strlen(Msg),i)==-1)
        {
            perror("mq_send error");
            return -1;
        }
    }
    mq_close(mqid);
    return 0;
}

讀出端

//psxmq_recv.c

#include <mqueue.h>
#include <stdbool.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>    
#include <time.h>      
#include <errno.h>
#include <string.h>

int main(int argc,char**argv)
{
    if(argc !=2)
    {
        printf("Usage: %s /mqname \r\n",argv[0]);
        return -1;
    }
    const int MAXSIZE =1024;
    char RecvBuff[MAXSIZE];
    int prio;
    ssize_t n;
    char* name  = argv[1];
    int   flags = O_RDONLY;

    memset(RecvBuff,0x00,sizeof(RecvBuff));
    mqd_t mqid = mq_open(name, flags);

    struct mq_attr attr;
    if(mqid==-1)
    {
        printf("error %s (%d)\r\n",strerror(errno),errno);
        return -1;
    }
    while(true)
    {
        if(mq_getattr(mqid,&attr)==-1)
        {
            printf("get attr error\r\n");
            break;
        }
        if(attr.mq_curmsgs==(long)0)
        {
            printf("no messages in queue\r\n");
            break;
        }

        if((n=mq_receive(mqid,RecvBuff,sizeof(RecvBuff),&prio))==-1)
        {
            perror("mq_receive error");
            return -1;
        }
        printf("read %ld bytes\r\n",(long)n);
        printf("prio is %d\r\n",prio);
        printf("%s \r\n",RecvBuff);
    }
    mq_close(mqid);
    mq_unlink(name);
    return 0;
}

Makefile

PROGS =Send Recv
CLEANFILES = core core.* *.core *.o temp.* *.out typescript* \
        *.lc *.lh *.bsdi *.sparc *.uw

all :${PROGS}


CFLAGS+=-g  
LIBS+=-lrt

Recv: psxmq_recv.o
    ${CC} ${CFLAGS} ${LIBPATH} $^  -o [email protected] ${LIBS}  
    @rm *.o

Send: psxmq_send.o
    ${CC} ${CFLAGS} ${LIBPATH} $^ -o [email protected]   ${LIBS}  
    @rm *.o

clean:
    rm -f ${PROGS} ${CLEANFILES}

3.參考