Linux 系统编程 ->进程通讯 -> 消息队列

masonzhang

  今天说下SYSV IPC 消息队列  消息队列简称为 MSG MSG是linux Kernel 维护的一个数据结构 相比管道它是广播式通讯 自由度更大 本文采用 通过类比windows 消息队列 学习 linux MSG  看下面的代码
    while (GetMessage(&msg, NULL, 0, 0))  
    {
        if (!TranslateAccelerator(msg.hwnd, hAccelTable, &msg))
        {
            TranslateMessage(&msg);
            DispatchMessage(&msg);
        }
    }
上面的代码给我一个思路 我们只要申请一个系统资源-> MSG 就可以组织多个进程通讯。优势很明显 也很方便 你如果用管道那必须按菊花链 or 星型组织进程 程序很复杂
  我们先解释下要使用的函数 重点只说前面2个
key_t ftok(const char *pathname, int proj_id);IPC_PRIVATE
int msgget(key_t key, int msgflg);
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);
int msgctl(int msqid, int cmd, struct msqid_ds *buf);
第一个函数顾名思义就是把文件做成一个KEY 解释一下 如何做
如指定文件的索引节点号为65538，换算成16进制为0x010002，而你指定的ID值为38，换算成16进制为0x26，则最后的key_t返回值为0x26010002。(--> 这段话来自一个网友的blog)
查询文件索引节点号的方法是： stat myfile
File: `myfile'
  Size: 0               Blocks: 0          IO Block: 4096   regular empty file
Device: 801h/2049d      Inode: 287830      Links: 1
Access: (0644/-rw-r--r--)  Uid: ( 1000/    xxx)   Gid: ( 1000/    xxx)
Access:  000000000 +0800
Modify: 000000000 +0800
Change:  000000000 +0800
chmod a-x myfile  
proj_id = the least significant 8 bits
key_t mykey=ftok("/my/private/idaho/myfile", 'a')
这个问题困于了我很久 最近才折腾明白 我们做这种 KEY 创造出来的队列 进程之间可以没有亲戚关系
下面下个参数 IPC_PRIVATE 当然和上面的情况相对立 这个用起来就很方便了
如何创建一个消息队列了
int msgget(key_t key, int msgflg);
键值 我们 说过了 下面说下 msgflg
你需要 man 5 ipc 看下 我贴一下 msgflg 的内容
  
0400    Read by user.
0200    Write by user.
0040    Read by group.
0020    Write by group.
0004    Read by others.
0002    Write by others
下面3个函数参考 demo 注意 收消息是堵塞的 如果不要堵塞需要设置标志位 IPC_NOWAIT
发消息是瞬发的
/*
*
* MSG DEMO
*
* Author: Masonzhang
*/
/* header define */
#include stdio.h>
#include stdlib.h>
#include string.h>
#include unistd.h>
#include sys/ipc.h>
#include sys/types.h>
#include sys/msg.h>
#include sys/wait.h>
#include signal.h>
/*end define */
/* macro define
* OCTAL :) man 5 IPC detail
* 0400 Read by user.
* 0200 Write by user.
* 0040 Read by group.
* 0020 Write by group.
* 0004 Read by others.
* 0002 Write by others.
********************************* */
#define MSG_R 0400
#define MSG_W 0200
/* end define */
/* data struct define */
struct {
        long m_type;
        char m_text[512];
}MSG={0,};
enum {first=0} ; // read first MSG record
enum {block=0} ; // block read MSG
/* end data struct define */
static void
Get_MSG(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg){
        char * MSG=(char *)msgp;
        bzero(MSG,msgsz);
        int msg_errno=msgrcv(msqid,msgp,msgsz,msgtyp,msgflg);
        MSG[msg_errno]='\0';
        if(strstr(MSG,"AT_OK") ) printf(" %ld child get copy is OK!\n",getpid());
}
static void
timer_handler(int int_val){
        static loop ;
        loop ++;
        if(loop >8) _exit(EXIT_FAILURE);
        else{
                loop %=10;
                printf("%d \n",int_val);
                sleep(1);
        }
        alarm(1);
}
int extern
main(int argc,char * arg[]){
        /* get qid from kernel */
        /* parent and child so using IPC_PRIVATE
         * msg flag setting R W
         */
        int qid=msgget(IPC_PRIVATE,MSG_R|MSG_W);
        if(qid==-1){
                perror(" MSG QUEUE CREATE ERROR \n ");
                exit(EXIT_FAILURE);
        }
        struct sigaction sa;
        sigemptyset(&sa.sa_mask);
        sa.sa_handler=timer_handler;
        if(sigaction(SIGALRM,&sa,NULL)) perror("signal install error"),exit(EXIT_FAILURE);
        int cpid=fork();
        if(cpid==0){
                alarm(1);
                int msg_errno;
                strcpy(MSG.m_text,"AT_OK!");
                if(msg_errno=msgsnd(qid,MSG.m_text,strlen(MSG.m_text),0)){
                        perror(" MSG DELIVERY FAILURE ");
                        _exit(0);
                }
                sleep(1);
                Get_MSG(qid,MSG.m_text,sizeof MSG.m_text,first,block); //block
                _exit(0);
        }
        /* error just game over */
        if(cpid 0) {
                perror(" CHILD CREATE FAILURE ");
                _exit(0);
        }
        /* parent process do job */
        if(cpid >0){
                int msg_errno=msgrcv(qid,MSG.m_text,sizeof MSG.m_text,first,block);
                if(msg_errno==-1){
                        perror(" MSG ACCEPT FAILURE \n");
                }
                MSG.m_text[msg_errno]='\0';
                if(strstr(MSG.m_text,"AT_OK") ) {
                        printf(" %ld parent get copy !\n",getpid());
                        while(1){
                                break;
                        }
                        bzero(MSG.m_text,sizeof MSG.m_text);
                        strcpy(MSG.m_text,"AT_OK!");
                        msgsnd(qid,&(MSG.m_text),strlen(MSG.m_text),0);
  
                }
                do {
                        int status;
                        int w = waitpid(cpid, &status, WUNTRACED | WCONTINUED );
                        if (w == -1) {
                                perror("WAITPID");
                                exit(EXIT_FAILURE);
                        }
                        if (WIFEXITED(status)) {
                                printf("exited, status=%d\n", WEXITSTATUS(status));
                        } else if (WIFSIGNALED(status)) {
                                printf("killed by signal %d\n", WTERMSIG(status));
                        } else if (WIFSTOPPED(status)) {
                                printf("stopped by signal %d\n", WSTOPSIG(status));
                        } else if (WIFCONTINUED(status)) {
                                printf("continued\n");
                        }
                } while (0);
                /* clean up */
                if(msgctl(qid,IPC_RMID,NULL)==-1){
                        perror(" CLEAN MSG ERROR");
                        exit(EXIT_FAILURE);
                }
        }
        exit(EXIT_SUCCESS);
}


本文来自ChinaUnix博客，如果查看原文请点：http://blog.chinaunix.net/u3/94090/showart_2117019.html