Posix消息隊列
消息隊列可以認為是一個消息鏈表，某個進程往一個消息隊列中寫入消息之前，不需要另外某個進程在該隊列上等待消息的達到，這一點與管道和FIFO相反。Posix消息隊列與System V消息隊列的區別如下：
1. 對Posix消息隊列的讀總是返回最高優先級的最早消息，對System V消息隊列的讀則可以返回任意指定優先級的消息。
2. 當往一個空隊列放置一個消息時，Posix消息隊列允許產生一個信號或啟動一個線程，System V消息隊列則不提供類似的機制。
Posix消息隊列操作函數如下：
#include <mqueue.h>
typedef int mqd_t;
mqd_t mq_open(const char *name, int oflag, ... /* mode_t mode, struct mq_attr *attr */);
返回： 成功時為消息隊列描述字，出錯時為－1。
功能： 創建一個新的消息隊列或打開一個已存在的消息的隊列。
#include <mqueue.h>
int mq_close(mqd_t mqdes);
返回： 成功時為0，出錯時為-1。
功能： 關閉已打開的消息隊列。
#include <mqueue.h>
int mq_unlink(const char *name)
返回： 成功時為0，出錯時為－1
功能： 從系統中刪除消息隊列。
#include <mqueue.h>
int mq_getattr(mqd_t mqdes, struct mq_attr *attr);
int mq_setattr(mqd_t mqdes, const struct mq_attr *attr, struct mq_attr *attr);
均返回：成功時為0， 出錯時為-1
每個消息隊列有四個屬性：
struct mq_attr
{
long mq_flags; /* message queue flag : 0, O_NONBLOCK */
long mq_maxmsg; /* max number of messages allowed on queue*/
long mq_msgsize; /* max size of a message (in bytes)*/
long mq_curmsgs; /* number of messages currently on queue */
};
每個消息均有一個優先級，它是一個小於MQ_PRIO_MAX的無符號整數
#define MQ_PRIO_MAX 32768
#include <mqueue.h>
int mq_send(mqd_t mqdes, const char *ptr, size_t len, unsigned int prio);
返回：成功時為0，出錯為－1
ssize_t mq_receive(mqd_t mqdes, char *ptr, size_t len, unsigned int *priop);
返回：成功時為消息中的字節數，出錯為－1
消息隊列的限制：
MQ_OPEN_MAX : 一個進程能夠同時擁有的打開著消息隊列的最大數目
MQ_PRIO_MAX : 任意消息的最大優先級值加1
#include <mqueue.h>
int mq_notify(mqd_t mqdes, const struct sigevent *notification);
返回： 成功時為0，出錯時為－1
功能： 給指定隊列建立或刪除異步事件通知
union sigval
{
int sival_int; /* Integer value */
void *sival_ptr; /* pointer value */
};
struct sigevent
{
int sigev_notify; /* SIGEV_{ NONE, ISGNAL, THREAD} */
int sigev_signo; /* signal number if SIGEV_SIGNAL */
union sigval sigev_value; /* passed to signal handler or thread */
void (*sigev_notify_function)(union sigval);
pthread_attr_t *sigev_notify_attribute;
};
異步信號安全函數
#include <signal.h>
int sigwait(const sigset_t *set, int *sig);
Posxi實時信號
信號可劃分為兩大小組：
1. 其值在SIGRTMIN和SIGRTMAX之間（包括兩者在內）的實時信號。
2. 所有其他信號：SIGALRM， SIGINT， SIGKILL等等。
void func(int signo, siginfo_t *info, void *context);
typedef struct
{
int si_signo; /* same value as signo argument */
int si_code; /* SI_{USER, QUEUE, TIMER, ASYNCIO, MESGQ}*/
union sigval si_value; /* integer or pointer value from sender */
} siginfo_t;
下面采用上面的函數，寫程序進程測試。
程序1：創建一個消息隊列，其名字是作為命令行參數指定，消息隊列創建成功後輸出隊列的屬性。程序如下：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include<sys/stat.h>
8 #define FILE_MODE (S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)
9
10 int main(int argc,char *argv[])
11 {
12 int c,flags;
13 mqd_t mqd;
14 struct mq_attr attr;
15
16 flags = O_RDWR|O_CREAT;
17 printf("create mqueue.\n");
18 while((c = getopt(argc,argv,"e")) != -1)
19 {
20 switch(c)
21 {
22 case ‘e‘:
23 flags |= O_EXCL;
24 break;
25 }
26 }
27 if(optind != argc-1)
28 {
29 perror("usage: mqcreate [-e] <name>");
30 exit(0);
31 }
32 if((mqd = mq_open(argv[optind],flags,FILE_MODE,NULL)) == -1)
33 {
34 perror("mq_open() error");
35 exit(-1);
36 }
37 mq_getattr(mqd,&attr);
38 printf("max #msgs = %ld,max #bytes/msg = %ld,#currently on queue = %ld\n",
39 attr.mq_maxmsg,attr.mq_msgsize,attr.mq_curmsgs);
40 mq_close(mqd);
41 exit(0);
42 }本測試是用的Linux Ubuntu系統，編譯程序的時候需要添加-lrt連接，如果不加會提示如下錯誤信息
因此正確編譯的方式如下：
程序編譯完成後，如果直接運行程序則提示mq_open失敗，提示mq_open permission denied。解決辦法是：
mkdir /dev/mqueue
mount -t mqueue none /dev/mqueue
然後再運行即可看到創建的消息隊列。程序結果如下所示：
程序2：練習mq_send和mq_receive函數，調用mqsend程序向消息隊列中寫入消息，調用mqreceive程序從消息隊列中讀取消息。程序如下所示：
mqsend程序：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include<sys/stat.h>
8 typedef unsigned int uint_t;
9
10 int main(int argc,char *argv[])
11 {
12 mqd_t mqd;
13 void *ptr;
14 size_t len;
15 uint_t prio;
16 if(argc != 4)
17 {
18 printf("usage: mqsend <name> <$bytes> <priority>\n");
19 exit(0);
20 }
21 len = atoi(argv[2]);
22 prio = atoi(argv[3]);
23 mqd = mq_open(argv[1],O_WRONLY);
24 ptr = calloc(len,sizeof(char));
25 if(mq_send(mqd,ptr,len,prio) == -1)
26 {
27 perror("mq_send() error:");
28 exit(-1);
29 }
30 exit(0);
31 }
mqreceive程序：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include<sys/stat.h>
8 typedef unsigned int uint_t;
9
10 int main(int argc,char *argv[])
11 {
12 int c,flags;
13 mqd_t mqd;
14 ssize_t n;
15 uint_t prio;
16 void *buff;
17 struct mq_attr attr;
18
19 flags = O_RDONLY;
20 while((c = getopt(argc,argv,"n")) != -1)
21 {
22 switch(c)
23 {
24 case ‘n‘:
25 flags |= O_NONBLOCK; //設置為非阻塞
26 break;
27 }
28 }
29 if(optind != argc-1)
30 {
31 printf("usage: mqreceive [-n] <name>");
32 exit(0);
33 }
34 mqd = mq_open(argv[optind],flags);
35 mq_getattr(mqd,&attr);
36 buff = malloc(attr.mq_msgsize);
37 if((n = mq_receive(mqd,buff,attr.mq_msgsize,&prio)) == -1)
38 {
39 perror("mq_receive error: ");
40 exit(-1);
41 }
42 printf("read %ld bytes,priority = %u\n",(long) n,prio);
43 exit(0);
44 }
程序執行結果如下所示：
程序3：信號通知函數使用，當有一個消息放置到某個空隊列中，該程序產生信號，通知進程消息隊列中放入了一個新的消息。程序如下：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include <signal.h>
8
9 typedef unsigned int uint_t;
10
11 volatile sig_atomic_t mqflag; //全局變量，檢查信號的產生
12 static void sig_usr1(int);
13
14 int main(int argc,char *argv[])
15 {
16 mqd_t mqd;
17 void *buff;
18 ssize_t n;
19 sigset_t zeromask,newmask,oldmask;
20 struct mq_attr attr;
21 struct sigevent sigev;
22 if(argc != 2)
23 {
24 printf("usage :mqnotify <name>");
25 exit(0);
26 }
27 mqd = mq_open(argv[1],O_RDONLY);
28 mq_getattr(mqd,&attr);
29 buff = malloc(attr.mq_msgsize);
30 sigemptyset(&zeromask);
31 sigemptyset(&newmask);
32 sigemptyset(&oldmask);
33 sigaddset(&newmask,SIGUSR1);
34 signal(SIGUSR1,sig_usr1);
35 sigev.sigev_notify = SIGEV_SIGNAL;
36 sigev.sigev_signo = SIGUSR1;
37 if(mq_notify(mqd,&sigev) == -1)
38 {
39 perror("mq_notify error");
40 exit(-1);
41 }
42 for(; ;)
43 {
44 sigprocmask(SIG_BLOCK,&newmask,&oldmask);
45 while(mqflag == 0)
46 sigsuspend(&zeromask); //掛起，等待
47 mqflag = 0;
48 mq_notify(mqd,&sigev);
49 n = mq_receive(mqd,buff,attr.mq_msgsize,NULL);
50 printf("read %ld bytes\n",(long) n);
51 sigprocmask(SIG_UNBLOCK,&newmask,NULL);
52 }
53 eixt(0);
54 }
55
56 static void sig_usr1(int signo)
57 {
58 mqflag = 1;
59 return ;
60 }
程序執行結果如下：
可以使用sigwait函數代替信號處理程序的信號通知，將信號阻塞到某個函數中，僅僅等待該信號的遞交。采用sigwait實現上面的程序如下：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include <signal.h>
8
9 int main(int argc,char *argv[])
10 {
11 mqd_t mqd;
12 int signo;
13 void *buff;
14 ssize_t n;
15 sigset_t newmask;
16 struct mq_attr attr;
17 struct sigevent sigev;
18 if(argc != 2)
19 {
20 printf("usage :mqnotify <name>");
21 exit(0);
22 }
23 mqd = mq_open(argv[1],O_RDONLY);
24 mq_getattr(mqd,&attr);
25 buff = malloc(attr.mq_msgsize);
26 sigemptyset(&newmask);
27 sigaddset(&newmask,SIGUSR1);
28 sigprocmask(SIG_BLOCK,&newmask,NULL);
29
30 sigev.sigev_notify = SIGEV_SIGNAL;
31 sigev.sigev_signo = SIGUSR1;
32 if(mq_notify(mqd,&sigev) == -1)
33 {
34 perror("mq_notify error");
35 exit(-1);
36 }
37 for(; ;)
38 {
39 sigwait(&newmask,&signo); //阻塞並等待該信號
40 if(signo == SIGUSR1)
41 {
42 mq_notify(mqd,&sigev);
43 while((n = mq_receive(mqd,buff,attr.mq_msgsize,NULL))>=0)
44 printf("read %ld bytes\n",(long) n);
45 if(errno != EAGAIN)
46 {
47 perror("mq_receive error");
48 exit(-1);
49 }
50 }
51 }
52 eixt(0);
53 }
啟動線程處理消息通知，程序如下：
1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <unistd.h>
4 #include <mqueue.h>
5 #include <fcntl.h>
6 #include <errno.h>
7 #include <signal.h>
8
9 mqd_t mqd;
10 struct mq_attr attr;
11 struct sigevent sigev;
12 static void notify_thread(union sigval);
13
14 int main(int argc,char *argv[])
15 {
16
17 if(argc != 2)
18 {
19 printf("usage :mqnotify <name>");
20 exit(0);
21 }
22 mqd = mq_open(argv[1],O_RDONLY | O_NONBLOCK);
23 mq_getattr(mqd,&attr);
24
25 sigev.sigev_notify = SIGEV_THREAD;
26 sigev.sigev_value.sival_ptr = NULL;
27 sigev.sigev_notify_function = notify_thread;
28 sigev.sigev_notify_attributes = NULL;
29
30 if(mq_notify(mqd,&sigev) == -1)
31 {
32 perror("mq_notify error");
33 exit(-1);
34 }
35 for(; ;)
36 {
37 pause();
38 }
39 eixt(0);
40 }
41 static void notify_thread(union sigval arg)
42 {
43 ssize_t n;
44 void *buff;
45 printf("notify_thread started\n");
46 buff = malloc(attr.mq_msgsize);
47 mq_notify(mqd,&sigev);
48 while((n = mq_receive(mqd,buff,attr.mq_msgsize,NULL))>=0)
49 printf("read %ld bytes\n",(long) n);
50 if(errno != EAGAIN)
51 {
52 perror("mq_receive error");
53 exit(-1);
54 }
55 free(buff);
56 pthread_exit(NULL);
57 }
Posix消息隊列