分类: LINUX
2012-01-06 21:20:32
一、消息队列的基本概念
消息队列 (也叫做报文队列)是Unix系统V版本中3种进程间通信机制之一。另外两种是信号灯和共享内存。这些IPC机制使用共同的授权方法。只有通过系统调用将标志符传递给核心之后,进程才能存取这些资源。这种系统IPC对象使用的控制方法和文件系统非常类似。使用对象的引用标志符作为资源表中的索引。 消息队列就是一个消息的链表。就是把消息看作一个记录,并且这个记录具有特定的格式以及特定的优先级。对消息队列有写权限的进程可以按照一定的规则添加新消息;对消息队列有读权限的进程则可以从消息队列中读出消息。 Linux采用消息队列的方式来实现消息传递。这种消息的发送方式是:发送方不必等待接收方检查它所收到的消息就可以继续工作下去,而接收方如果没有收到消息也不需等待。这种通信机制相对简单,但是应用程序使用起来就需要使用相对复杂的方式来应付了。新的消息总是放在队列的末尾,接收的时候并不总是从头来接收,可以从中间来接收。 消息队列是随内核持续的并和进程相关,只有在内核重起或者显示删除一个消息队列时,该消息队列才会真正被删除。因此系统中记录消息队列的数据结构 (struct ipc_ids msg_ids)位于内核中,系统中的所有消息队列都可以在结构msg_ids中中找到访问入口。 IPC标识符:每一个I P C目标都有一个唯一的I P C标识符。这里所指的I P C目标是指一个单独的消息队列、一个信号量集或者一个共享的内存段。系统内核使用此标识符在系统内核中指明 I P C目标。 IPC 关键字:想要获得唯一的标识符,则必须使用一个 I P C关键字。客户端进程和服务器端进程必须双方都同意此关键字。这是建立一个客户机/服务器框架的第一步。在System V IPC机制中,建立两端联系的路由方法是和I P C关键字直接相关的。通过在应用程序中设置关键字值,每一次使用的关键字都可以是相同的。一般情况下,可以使用f t o k ( )函数为客户端和服务器端产生关键字值。 命令ipcs用于读取System V IPC目标的状态。 下面是ipcs命令输出的例子: [root@wanglong wanglong]# ipcs —— Shared Memory Segments ——– key shmid owner perms bytes nattch status 0×00000000 0 root 644 40 2 0×00000000 32769 root 644 16384 2 0×00000000 65538 root 644 268 2 —— Semaphore Arrays ——– key semid owner perms nsems 0×000000a7 0 root 600 1 0×00000000 98305 apache 600 1 0×00000000 65538 apache 600 1 0×00000000 131075 apache 600 1 0×00000000 163844 apache 600 1 0×00000000 196613 apache 600 1 0×00000000 229382 apache 600 1 0×00000000 262151 apache 600 1 0×00000000 294920 apache 600 1 —— Message Queues ——– key msqid owner perms used-bytes messages 内核中实现消息传递机制的代码基本上都在文件ipc/msg.c中,消息队列的主要调用有下面4个,这里只作简单介绍: (1)msgget:调用者提供一个消息队列的键标 (用于表示个消息队列的唯一名字),当这个消息队列存在的时候, 这个消息调用负责返回这个队列的标识号;如果这个队列不存在,就创建一个消息队列,然后返回这个消息队列的标识号 ,主要由sys_msgget执行。 (2)msgsnd:向一个消息队列发送一个消息,主要由sys_msgsnd执行。 (3)msgrcv:从一个消息队列中收到一个消息,主要由sys_msgrcv执行。 (4)msgctl:在消息队列上执行指定的操作。根据参数的不同和权限的不同,可以执行检索、删除等的操作,主要由sys_msgctl执行。 下面的例子很好的演示了创建、发送、读取、改变权限以及删除消息队列各种操作: #include struct mymsgbuf { void send_message(int qid, struct mymsgbuf *qbuf, long type, char *text); if(argc == 1) /* Create unique key via call to ftok() */ /* Open the queue - create if necessary */ { switch(tolower(argv[1][0])) default: usage(); } return(0); void send_message(int qid, struct mymsgbuf *qbuf, long type, char *text) if((msgsnd(qid, (struct msgbuf *)qbuf, void read_message(int qid, struct mymsgbuf *qbuf, long type) printf(“Type: %ld Text: %sn”, qbuf->mtype, qbuf->mtext); void remove_queue(int qid) void change_queue_mode(int qid, char *mode) /* Get current info */ /* Convert and load the mode */ /* Update the mode */ void usage(void) 程序保存为 ipcs.c 编译:gcc -o ipcs ipcs.c 程序运行结果解释: [root@wanglong wanglong]# ./ipcs s 001 hello! Sending a message … [root@wanglong wanglong]# ./ipcs s 001 world! Sending a message … [root@wanglong wanglong]# ./ipcs s 001 you Sending a message … [root@wanglong wanglong]# ./ipcs s 001 and Sending a message … [root@wanglong wanglong]# ./ipcs s 001 me! Sending a message … [root@wanglong wanglong]# ./ipcs r 001 Reading a message … Type: 001 Text:hello! [root@wanglong wanglong]# ./ipcs r 001 Reading a message … Type: 001 Text:world! [root@wanglong wanglong]# ./ipcs r 001 Reading a message … Type: 001 Text:you [root@wanglong wanglong]# ./ipcs d 001 /*删除了消息队列001*/ [root@wanglong wanglong]# ./ipcs r 001 Reading a message .. ./* 因为删除了,所以读不出消息了*/
ipcs -q: 只显示消息队列。
ipcs -s: 只显示信号量。
ipcs -m: 只显示共享内存。
ipcs –help: 其他的参数。
#include
#include
#include
#include
#include
#include
#define MAX_SEND_SIZE 80
long mtype;
char mtext[MAX_SEND_SIZE];
};
void read_message(int qid, struct mymsgbuf *qbuf, long type);
void remove_queue(int qid);
void change_queue_mode(int qid, char *mode);
void usage(void);
int main(int argc, char *argv[])
{
key_t key;
int msgqueue_id;
struct mymsgbuf qbuf;
usage();
key = ftok(“.”, ’m');
if((msgqueue_id = msgget(key, IPC_CREAT|0660)) == -1)
perror(“msgget”);
exit(1);
}
{
case ’s’: send_message(msgqueue_id, (struct mymsgbuf *)&qbuf,atol(argv[2]), argv[3]);
break;
case ’r': read_message(msgqueue_id, &qbuf, atol(argv[2]));
break;
case ’d': remove_queue(msgqueue_id);
break;
case ’m': change_queue_mode(msgqueue_id, argv[2]);
break;
}
{
/* Send a message to the queue */
printf(“Sending a message …n”);
qbuf->mtype = type;
strcpy(qbuf->mtext, text);
strlen(qbuf->mtext)+1, 0)) ==-1)
{
perror(“msgsnd”);
exit(1);
}
}
{
/* Read a message from the queue */
printf(“Reading a message …n”);
qbuf->mtype = type;
msgrcv(qid, (struct msgbuf *)qbuf, MAX_SEND_SIZE, type, 0);
}
{
/* Remove the queue */
msgctl(qid, IPC_RMID, 0);
}
{
struct msqid_ds myqueue_ds;
msgctl(qid, IPC_STAT, &myqueue_ds);
sscanf(mode, ”%ho”, &myqueue_ds.msg_perm.mode);
msgctl(qid, IPC_SET, &myqueue_ds);
}
{
fprintf(stderr, ”msgtool - A utility for tinkering with msg queuesn”);
fprintf(stderr, ”nUSAGE: msgtool (s)end n”);
fprintf(stderr, ” (r)ecv n”);
fprintf(stderr, ” (d)eleten”);
fprintf(stderr, ” (m)ode n”);
exit(1);
}