实现posix消息队列示例分享

mqueue.h

#ifndef __PS_MQUEUE_H
#define __PS_MQUEUE_H

#include <unistd.h>
#include <sys/types.h>

typedef struct mq_info     *mqd_t;
typedef struct mq_attr    mq_attr;

#ifdef __cplusplus
extern "C" {
#endif

    mqd_t   mq_open(const char *name, int flag, .../*mode_t mode, struct mq_attr *attr*/);
    int     mq_close(mqd_t mqdes);
    int     mq_unlink(const char *name);

    int     mq_getattr(mqd_t mqdes,mq_attr *attr);
    int     mq_setattr(mqd_t mqdes,const mq_attr *attr,mq_attr *old);

    int     mq_send(mqd_t mqdes,const char *ptr,size_t len,unsigned int prio);
    int     mq_receive(mqd_t mqdes,char *ptr,size_t len,unsigned int *priop);

   
    //
    void    mq_info_test(mqd_t mqdes);

#ifdef __cplusplus
}
#endif
#endif
</div>

多进程，多线程创建同一个队列测试

void *create_mq(void *name){
    mqd_t mq;
    mq = mq_open("/tmp/mqfile", O_CREAT,FILE_MODE,0);

    if (mq == (mqd_t) -1) {
        err_ret(errno, "mq_open() error");
        return 0;
    }

    mq_info_test(mq);

    mq_close(mq);

    return 0;
}

int main(){

    mq_unlink("/tmp/mqfile");

    if (Fork() == 0) {
        create_mq("/tmp/mqfile");
        exit(0);
    }

    Create_detach_thread(create_mq, "/tmp/mqfile");
    Create_detach_thread(create_mq, "/tmp/mqfile");

    sleep(50);

    //mq_unlink("/tmp/mqfile");

    return 0;
}
</div>

测试结果

属性设置、获取测试

void print_attr(mq_attr *attr){
    assert(attr);

    err_msg(" mq_attr mq_flag:0x%0x"
            " mq_curmsgs:%d"
            " mq_msgsize:%d"
            " mq_maxmsg:%d"
            ,attr->mq_flags
            ,attr->mq_curmsgs
            ,attr->mq_msgsize
            ,attr->mq_maxmsg);
}

void *create_mq(void *name){
    pthread_t tid;
    mq_attr attr,old;
    mqd_t mq;
    int flag;

    flag = O_CREAT;

    tid = pthread_self();

    if ((long)tid % 2 != 0) {
        flag = O_NONBLOCK;
    }

    mq = mq_open("/tmp/mqfile", flag | O_CREAT,FILE_MODE,0);

    if (mq == (mqd_t) -1) {
        err_ret(errno, "mq_open() error");
        return 0;
    }

    if ((long)tid % 2 == 0) {
        attr.mq_flags = O_NONBLOCK;
        mq_setattr(mq, &attr, &old);
    }
    else
        mq_getattr(mq, &old);

    print_attr(&old);

    //mq_info_test(mq);

    mq_close(mq);

    return 0;
}

int main(){
    pid_t pid;

    mq_unlink("/tmp/mqfile");

    if ((pid=Fork()) == 0) {
        create_mq("/tmp/mqfile3");
        Create_detach_thread(create_mq, "/tmp/mqfile1");
        Create_detach_thread(create_mq, "/tmp/mqfile2");
        sleep(1);
        exit(0);
    }

    Create_detach_thread(create_mq, "/tmp/mqfile1");
    Create_detach_thread(create_mq, "/tmp/mqfile2");
    create_mq("/tmp/mqfile3");

    wait(0);

    sleep(5);

    //mq_unlink("/tmp/mqfile");

    return 0;
}
</div>

测试注册通知规则

int main(){
    pid_t pid;
    Init_wait();
    mqd_t mq;

    sigevent_t sige;

    mq_unlink("/tmp/mqfile");
    mq = mq_open("/tmp/mqfile", O_CREAT,FILE_MODE,0);

    Signal(SIGCHLD, SIG_DFL);

 &n