Linuxc - 消息队列

摘要：用于控制当前消息队列满或队列消息到达系统范围的限制时将要发生的事情返回值成功，消息数据的一分副本将被放到消息队列中，并返回，失败时返回用来从一个消息队列获取消息是由函数返回的消息队列标识符。

​ --详情见《unix环境高级编程》15.7小节

​ 在《unix环境高级编程》15.7小节最后，总结了消息队列的优缺点。结论是，在新的应用程序中不应当再使用它们。确实是这样，追求高效，我们可以用共享内存，追求方便我们可以用管道或者UNIX域，消息队列就是处于这种尴尬的位置。

#include 
/*    用来创建和访问一个消息队列
    key:  与其他的IPC机制一样，程序必须提供一个键来命名某个特定的消息队列
    msg： flg是一个权限标志，表示消息队列的访问权限，它与文件的访问权限一样
    返回值：返回一个以key命名的消息队列的标识符（非零整数），失败时返回-1.
*/
int msgget(key_t  key, int msgflg);
/*    用来把消息添加到消息队列中
    msgid：是由msgget函数返回的消息队列标识符。
    msg_ptr：msg_ptr是一个指向准备发送消息的指针
    msg_sz：是msg_ptr指向的消息的长度，注意是消息的长度，而不是整个结构体的长度，也就是说msg_sz是不包括长整型消息类型成员变量的长度。
    msgflg：用于控制当前消息队列满或队列消息到达系统范围的限制时将要发生的事情
    返回值： 成功，消息数据的一分副本将被放到消息队列中，并返回0，失败时返回-1
*/
int msgsnd(int msgid, const void *msg_ptr, size_t msg_sz, int msgflg); 
/*用来从一个消息队列获取消息
    msgid：是由msgget函数返回的消息队列标识符。
    msg_ptr：msg_ptr是一个指向准备发送消息的指针
    msg_st： 同上
    msgtype：可以实现一种简单的接收优先级。如果msgtype为0，就获取队列中的第一个消息。如果它的值大于零，将获取具有相同消息类型的第一个信息。如果它小于零，就获取类型等于或小于msgtype的绝对值的第一个消息。
    msgflg：用于控制当队列中没有相应类型的消息可以接收时将发生的事情
    返回值： 成功时，该函数返回放到接收缓存区中的字节数，消息被复制到由msg_ptr指向的用户分配的缓存区中，然后删除消息队列中的对应消息。失败时返回-1.
*/
int msgrcv(int msgid, void *msg_ptr, size_t msg_st, long int msgtype, int msgflg);
/*    用来控制消息队列，它与共享内存的shmctl函数相似
    msgid：是由msgget函数返回的消息队列标识符。
    Command：是将要采取的动作，它可以取3个值，
        IPC_STAT：把msgid_ds结构中的数据设置为消息队列的当前关联值，即用消息队列的当前关联值覆盖msgid_ds的值。
        IPC_SET：如果进程有足够的权限，就把消息列队的当前关联值设置为msgid_ds结构中给出的值
        IPC_RMID：删除消息队列
    buf：是指向msgid_ds结构的指针，它指向消息队列模式和访问权限的结构。msgid_ds结构至少包括以下成员：
        struct msgid_ds {  
            uid_t shm_perm.uid;  
            uid_t shm_perm.gid;  
            mode_t shm_perm.mode;  
        };
    返回值： 成功时返回0，失败时返回-1.
*/
int msgctl(int msgid, int command, struct msgid_ds *buf);

client和server的通信

//comm.h
#ifndef _COMM_H_
#define _COMM_H_

#include 
#include 
#include 
#include 
#include 
#include 

struct msgbuf
{
    long mtype;
    char mtext[1024];
};

#define SERVER_TYPE 1
#define CLIENT_TYPE 2

int createMsgQueue();
int getMsgQueue();
int destoryMsgQueue(int msg_id);
int sendMsgQueue(int msg_id, int who, char* msg);
int recvMsgQueue(int msg_id, int recvType, char out[]);

#endif

//comm.c
#include "comm.h"

static int commMsgQueue(int flags)
{
    key_t key = ftok("/tmp", 0x6666);
    if(key < 0)
    {
        perror("ftok");
        return -1;
    }

    int msg_id = msgget(key, flags);
    if(msg_id < 0)
    {
        perror("msgget");
    }
    return msg_id;
}

int createMsgQueue()
{
    return commMsgQueue(IPC_CREAT|IPC_EXCL|0666);
}

int getMsgQueue()
{
    return commMsgQueue(IPC_CREAT);
}

int destoryMsgQueue(int msg_id)
{
    if(msgctl(msg_id, IPC_RMID, NULL) < 0)
    {
        perror("msgctl");
        return -1;
    }
    return 0;
}

int sendMsgQueue(int msg_id, int who, char* msg)
{
    struct msgbuf buf;
    buf.mtype = who;
    strcpy(buf.mtext, msg);

    if(msgsnd(msg_id, (void*)&buf, sizeof(buf.mtext), 0) < 0)
    {
        perror("msgsnd");
        return -1;
    }
    return 0;
}

int recvMsgQueue(int msg_id, int recvType, char out[])
{
    struct msgbuf buf;
    int size=sizeof(buf.mtext);
    if(msgrcv(msg_id, (void*)&buf, size, recvType, 0) < 0)
    {
        perror("msgrcv");
        return -1;
    }

    strncpy(out, buf.mtext, size);
    out[size] = 0;
    return 0;
}

//server.c
#include "comm.h"

int main()
{
    int msgid = createMsgQueue();

    char buf[1024] = {0};
    while(1)
    {
        recvMsgQueue(msgid, CLIENT_TYPE, buf);
        if(strcasecmp("quit", buf) == 0)
            break;
        printf("client# %s
", buf);

        printf("Please enter# ");
        fflush(stdout);
        ssize_t s = read(0, buf, sizeof(buf));
        if(s>0)
        {
            buf[s-1]=0;
            sendMsgQueue(msgid, SERVER_TYPE, buf);
            printf("send done, wait recv...
");
        }
    }

    destoryMsgQueue(msgid);
    return 0;
}

//client.c
#include "comm.h"

int main()
{
    int msgid = getMsgQueue();

    char buf[1024] = {0};
    while(1)
    {
        printf("Please Enter# ");
        fflush(stdout);
        ssize_t s = read(0, buf, sizeof(buf));
        if(s > 0)
        {
            buf[s-1]=0;
            sendMsgQueue(msgid, CLIENT_TYPE, buf);
            if(strcasecmp("quit", buf) == 0)
                break;
            printf("send done, wait recv...
");
        }

        recvMsgQueue(msgid, SERVER_TYPE, buf);
        printf("server# %s
", buf);
    }
    return 0;
}

运行结果

root@jonathan-pc:~/test/msg# ls
client.c  comm.c  comm.h  server.c
root@jonathan-pc:~/test/msg# gcc client.c comm.c -o c
root@jonathan-pc:~/test/msg# gcc server.c comm.c -o s
root@jonathan-pc:~/test/msg# ./s
client# hello
Please enter# nihao
send done, wait recv...
root@jonathan-pc:~/test/msg#

root@jonathan-pc:~/test/msg# ./c
Please Enter# hello
send done, wait recv...
nihaoserver# nihao
Please Enter# quit
send done, wait recv...