IPC(SystemV) 之 消息队列

标签：ipc   msgqueue   

本文介绍最后一个SystemV的IPC，即消息队列。消息队列提供了一个从一个进程向另外一个进程发送消息的方式。而且每个数据块都被认为包含一个类型，所以接收进程可以独立的接收含有不同类型的数据块。在实际的使用中,还是要结合使用场景来决定是否使用消息队列.我遇到过的场景就是进程之间需要有序的去共享具有类型的消息.除此之外我也提供不了一个更贴切的使用场景,可能大多数人喜欢用Unix域套接字或者共享内存来代替这种方式.

下面详细介绍一下共享内存.

首先是头文件,和另外两个IPC类似,消息队列的头文件是<sys/msg.h>,同样可能需要额外包含<sys/types.h> 和 <sys/ipc.h>

函数如下:

int msgctl(int msg_id, int command, struct msg_ds *buf);

int msgget(key_t key, int flag);

int msgrcv(int msg_id, void *msg_ptr, size_t msg_sz, long int msgtype, int msgflag);

int msgsnd(int msg_id, void * msg_ptr, size_t msg_sz, int msgflag);

1. msgget 创建一个消息队列的唯一标记,供其它函数调用.当然可以在key使用IPC_PRIVATE,指定只能在进程内使用.flag可选的值低九位为消息队列的权限.其它标记有IPC_CREAT 创建或者获取一个已经存在的消息队列.如果接合IPC_EXCL.则只允许创建,如果消息队列已经存在,那么返回错误.

msgget成功返回消息队列的id, 失败的话返回-1

2. msgctl  消息队列控制函数,第一个参数就是msgget返回的id, 第二个参数制定对消息队列的操作,大多数时间,我们只要记住IPC_RMID就行了.其它的我美柚怎么用过.比如可以通过IPC_STAT,IPC_SET获取到或者设置消息队列的状态.后面的buf为消息队列的一些属性.

msgctl成功调用返回0, 失败返回-1

3. msgrcv  顾名思义啊,就是收取一段消息.第一个参数都是msgget返回的id, 第二个参数是一个消息结构.在介绍完这个函数后,我们见识一下这个结构. 第三个参数msg_sz, 即第二个参数的大小.第四个参数,指定接收数据的类型.不过并非只有这种用法.msg_type实现了一个简易的优先级以及过滤方式.如果大于0, 那么只能接受msg_type指定的值,如果等于0, 则按顺序接受.如果小于0, 则接收小于等于msg_type绝对值的消息类型.最后一个参数flag用来控制如果没有消息的时候, 进程采取什么样的动作.可选的标记有IPC_NOWAIT,如果制定了这个标记,则在没有消息的时候进程直接退出,如果没有指定这个消息.那么进程会一直等待.

msgrcv成功返回接收缓冲区的字节数, 失败返回-1

4. msgsnd 发送消息.第一个参数msg_id, 通过msgget获取到的消息队列标记,第二个参数msg_ptr,将要发送的内容.第三个参数msg_sz, 指定消息的长度. 第四个参数msgflag,指定了消息队列满或者达到系统的指定上限的时候将要发生的事情.如果指定了IPC_NOWAIT,则不等待直接返回-1, 如果不指定这个选项,则会阻塞.

msgsnd 成功返回写入缓冲区的字节数,失败返回-1 .

接下来说一下消息体结构,其实就是一个结构体,区别就是前四个字节必须是一个long int类型制定消息类型.

struct msg

{

long int msg_type;

void *buf;   //这里的内容可以随意发挥

}

ok,下面我们设计一个场景.三个进程之间相互聊天.谁拿到信号量谁说话.没有拿到信号量的进程则等待接受消息.

还是总结一下遇到的问题吧.我以为我会了,但是还是有很多细节问题需要注意.

1. msgsnd函数的第三个参数是消息体的长度,不包括类型.我开始的时候sizeof(struct msg_struct).结果肯定是不行的.后来看到了这一条约束以后,我减去了一个sizeof(int).发现还是不行.最后才留意到.不是全部长度,不是int长度,而是long int.这样不能怨我,刚开始的时候我在32位机器上面编译的程序.没有注意到这一点,也没有遇到问题.但是这两天换了系统,换成了64位的系统.所以才遇到了这个问题.很是郁闷.

2. scanf.学习了这么就的C语言,才发现了我到现在页不会用scanf.这玩意实在太坑爹了.我明明输入了内容.但就是没有输出.多半就是和缓冲区有关系.具体没有去深究.直接换了函数,改用fgets.问题解决.

3. 因为是几乎相同的代码,所以在释放资源上面,页出现了问题,第一个进程退出,删除了消息队列,第二个进程再去读的时候就会返回错误.以至于第二个进程无法释放.

综上,修改后的代码如下:

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/sem.h>
#include <sys/msg.h>

#define MSG_LEN 1024

#define MSG_TYPE 1

union semun {
    int              val;    /* Value for SETVAL */
    struct semid_ds *buf;    /* Buffer for IPC_STAT, IPC_SET */
    unsigned short  *array;  /* Array for GETALL, SETALL */
    struct seminfo  *__buf;  /* Buffer for IPC_INFO
                                (Linux-specific) */
};

struct msg_buffer
{
    long int msg_type;
    char buffer[MSG_LEN];
};

int main(int argc, char ** argv)
{
    int ret;
    int quit_talking = 0;
    int sem_id, msg_id;

    struct sembuf sem_p;
    struct sembuf sem_v;
    union semun option;

    struct msg_buffer msg;

    msg_id = msgget((key_t)123456, 0666 | IPC_CREAT);
    if (msg_id < 0)
    {
        perror("create msg failed\n");
        exit(EXIT_FAILURE);
    }

    sem_id = semget((key_t)123478, 1, 0666 | IPC_CREAT);
    if (sem_id < 0)
    {
        perror("sem create failed\n");
        exit(EXIT_FAILURE);
    }

    option.val = 1;   //注1
    ret = semctl(sem_id, 0, SETVAL, option);
    if (ret < 0)
    {
        perror("semctl failed\n");
        exit(EXIT_FAILURE);
    }

    sem_p.sem_num = 0;
    sem_p.sem_op = -1;
    sem_p.sem_flg = SEM_UNDO;

    sem_v.sem_num = 0;
    sem_v.sem_op = 1;
    sem_v.sem_flg = SEM_UNDO;

    while (quit_talking == 0)
    {
        semop(sem_id, &sem_p, 1);

        bzero(&msg, sizeof(struct msg_buffer));
        ret = msgrcv(msg_id, &msg, sizeof(struct msg_buffer) - sizeof(long int ), 1, IPC_NOWAIT);
        if (ret > 0)
        {
            printf("msgtype is %ld, msg is %s\n", msg.msg_type, msg.buffer);

            if (strncmp(msg.buffer, "end", 3) == 0)
            {
                quit_talking = 1;
                continue;
            }
            bzero(&msg, sizeof(struct msg_buffer));
        }

        printf("please input your msg\n");
        fgets(msg.buffer, MSG_LEN, stdin);
        msg.msg_type = MSG_TYPE;
        
        msgsnd(msg_id, &msg, strlen(msg.buffer), IPC_NOWAIT);

        if (strncmp(msg.buffer, "end", 3) == 0)
        {
            quit_talking = 1;
        }

        semop(sem_id, &sem_v, 1);
    }

    ret = semctl(sem_id, 0, IPC_RMID, option);
    if (ret < 0)
    {
       printf("sem rm failed\n");
       exit(EXIT_FAILURE);
    }

    sleep(1);
    ret = msgctl(msg_id, IPC_RMID, NULL);
    if (ret < 0)
    {
        printf("msg rm failed\n");
        exit(EXIT_FAILURE);
    }

    return 0;
}

本来想写一个稍微复杂一点的例子.不过想了半天,不是很好写,并且也不利于理解.所以就写了一个相对简单的.结果如下:

IPC(SystemV) 之 消息队列

标签：ipc   msgqueue   

原文地址：http://blog.csdn.net/cp3alai/article/details/44565429