linux开发---24.【C】进程间通信之posix消息队列

1.


2. mq的send与receive
一个进程调用mq_send发送消息
一个进程调用mq_receive接收消息
mq_send完消息之后,mq_receive可以在任何时间去接收消息
2.1 mq_send

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

int main ( int argc, char *argv[] )
{
    int i;
    int oflag;
    mqd_t mqd;
    struct mq_attr attr;
    char buf[1024];

    oflag = O_CREAT|O_RDWR;
    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", oflag, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    for(i=0; i<10; i++)
    {
        snprintf(buf, 10, "mq_send=%d\n", i);
        mq_send(mqd, buf, 10, 3);
    }
    mq_close(mqd);
    return EXIT_SUCCESS;
}

2.2 mq_recv

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

int main ( int argc, char *argv[] )
{
    int i;
    int oflag;
    int prio;
    mqd_t mqd;
    struct mq_attr attr;
    char* buf = NULL;

    oflag = O_CREAT|O_RDWR;
    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", oflag, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    //这儿的buf为什么不能预先分配呢？或者用数组呢？
    //因为mq_receive的len参数不能小于mq_msgsize
    //在没有调用mq_getattr之前是不知道mq_msgsiz是多少的，所以需要这儿动态分配
    buf = calloc(attr.mq_msgsize,sizeof(char));
    for(i=0; i<attr.mq_curmsgs; i++)
    {
        mq_receive(mqd, buf, attr.mq_msgsize, &prio);      //注意len参数不能小于能加到所指定队列中的消息的最大大小(mq_msgsize)
        printf("recv:buf=%s,prio=%d\n", buf, prio);
    }
    free(buf);
    mq_close(mqd);
    return EXIT_SUCCESS;
}

3.使用notify
使用步骤:
    a.运行mq_create生成一个mq
   b. 运行mq_receive等侍接收
   c. 运行mq_send发送消息
3.1 加入mq_create

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

int main ( int argc, char *argv[] )
{
    int i;
    int oflag;
    mqd_t mqd;
    struct mq_attr attr;
    char buf[1024];

    oflag = O_CREAT|O_RDWR;
    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", oflag, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("create:mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    mq_close(mqd);
    return EXIT_SUCCESS;
}

执行时需要选运行一次mq_create建立mq
3.2 mq_send不变

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

int main ( int argc, char *argv[] )
{
    int i;
    int oflag;
    mqd_t mqd;
    struct mq_attr attr;
    char buf[1024];

    oflag = O_CREAT|O_RDWR;
    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", oflag, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    printf("MQ_OPEN_MAX=%ld,MQ_PRIO_MAX=%ld\n", sysconf(_SC_MQ_OPEN_MAX), sysconf(_SC_MQ_PRIO_MAX));
    for(i=0; i<10; i++)
    {
        snprintf(buf, 10, "mq_send=%d\n", i);
        mq_send(mqd, buf, 10, 3);
    }
    mq_close(mqd);
    return EXIT_SUCCESS;
}

3.3 重点mq_receive

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>
#include <signal.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

mqd_t mqd;
struct mq_attr attr;

static void sig_usr1(int signo)
{
    int i;
    int prio;
    char* buf = NULL;
    buf = calloc(attr.mq_msgsize,sizeof(char));
    for(i=0; i<attr.mq_curmsgs; i++)
    {
        mq_receive(mqd, buf, attr.mq_msgsize, &prio);
        printf("recv:buf=%s,prio=%d\n", buf, prio);
    }
    free(buf);
}

int main ( int argc, char *argv[] )
{
    struct sigevent sigev;

    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", O_RDWR, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    signal(SIGUSR1, sig_usr1);
    sigev.sigev_notify = SIGEV_SIGNAL;
    sigev.sigev_signo = SIGUSR1;
    mq_notify(mqd, &sigev);
    for(;;)
        pause();
    //mq_close(mqd);
    return EXIT_SUCCESS;
}

3.3.1 改进
a.这儿有个问题: 在运行mq_receive之前,mq不为空,调用mq_recv-->调用mq_send时，
mq_recv不会收到任何singal

cong@msi:/tmp$ man mq_notify
Message notification occurs only when a new message arrives and the queue was previously empty.
注意:mq_notify产生的条件:队列由空变有数据才会产生一个mq_notify

b.mq_notify只会产生一次信号，
mq_create-->mq_recv-->mq_send 正常，但再次调用mq_send时就不会再产生信号了

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h> /* For O_* constants */
#include <sys/stat.h> /* For mode constants */
#include <mqueue.h>
#include <signal.h>

#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH)

mqd_t mqd;
struct mq_attr attr;
int read_mq_msg()
{
    int i;
    int prio;
    char* buf = NULL;
    buf = calloc(attr.mq_msgsize,sizeof(char));
    for(i=0; i<attr.mq_curmsgs; i++)
    {
        mq_receive(mqd, buf, attr.mq_msgsize, &prio);
        printf("recv:buf=%s,prio=%d\n", buf, prio);
    }
    free(buf);
}

static void sig_usr1(int signo)
{
    printf("recv signal, and then read\n");
    read_mq_msg();                            //收到signal之后读空队列
    mq_notify_register();                     //并再次等侍接收signal
}

int mq_notify_register()
{
    struct sigevent sigev;
    signal(SIGUSR1, sig_usr1);
    sigev.sigev_notify = SIGEV_SIGNAL;
    sigev.sigev_signo = SIGUSR1;
    mq_notify(mqd, &sigev);
}

int main ( int argc, char *argv[] )
{
    //mqd_t mq_open(const char *name, int oflag, mode_t mode, struct mq_attr *attr);
    mqd = mq_open("/hellomq", O_RDWR, FILE_MODE, NULL);
    if(mqd < 0)
    {
        perror("mq_open error");
        return -1;
    }
    mq_getattr(mqd, &attr);
    printf("mq_maxmsg=%ld, mq_msgsize=%ld, mq_curmsgs=%ld\n", attr.mq_maxmsg, attr.mq_msgsize, attr.mq_curmsgs);
    if(attr.mq_curmsgs!=0)
    {
        printf("mq_curmsgs is not empty, read it \n");
        read_mq_msg();                 //先把队列中的数据读空
    }
    mq_notify_register();             //产生首次signal
    for(;;)
        pause();
    //mq_close(mqd);
    return EXIT_SUCCESS;
}

4.Makefile

EXE=mq_create mq_send mq_recv
CC=gcc
SRC=$(wildcard *.c)
#OBJ=$(SRC:.c=.o)
OBJ=$(patsubst %.c,%.o,$(SRC))
DEP=$(patsubst %.c,.%.d,$(SRC))
CFLAGS=-g -O0
ALL:$(EXE)
mq_create:mq_create.o
    $(CC) $^ -o $@ -lrt
mq_send:mq_send.o
    $(CC) $^ -o $@ -lrt
mq_recv:mq_recv.o
    $(CC) $^ -o $@ -lrt              ###这儿都要加上-lrt

$(DEP):.%.d:%.c
    @set -e; rm -f $@; \
    $(CC) -MM  $< > $@.$$$$; \
    sed 's,/($*/)/.o[ :]*,/1.o $@ : ,g' < $@.$$$$ > $@; \
    rm -f $@.$$$$

-include $(DEP)
clean:
    @rm $(EXE) $(OBJ) $(DEP) *.o -f

5.源码打包 
3mq_notify.rar(下载后改名为3mq_notify.tar.gz)