【发布时间】：2011-03-04 15:12:54
【问题描述】：
如何在基于 Linux 的系统上的 c 程序中使用 mqueue（消息队列）？
我正在寻找一些好的代码示例，这些示例可以展示如何以正确和适当的方式完成此操作，也许是一个操作指南。
【问题讨论】：
【发布时间】：2011-03-04 15:12:54
【问题描述】：
以下是一个简单的服务器示例，它接收来自客户端的消息，直到它收到一条“退出”消息告诉它停止。
服务器的代码：
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>
#include <mqueue.h>
#include "common.h"
int main(int argc, char **argv)
{
mqd_t mq;
struct mq_attr attr;
char buffer[MAX_SIZE + 1];
int must_stop = 0;
/* initialize the queue attributes */
attr.mq_flags = 0;
attr.mq_maxmsg = 10;
attr.mq_msgsize = MAX_SIZE;
attr.mq_curmsgs = 0;
/* create the message queue */
mq = mq_open(QUEUE_NAME, O_CREAT | O_RDONLY, 0644, &attr);
CHECK((mqd_t)-1 != mq);
do {
ssize_t bytes_read;
/* receive the message */
bytes_read = mq_receive(mq, buffer, MAX_SIZE, NULL);
CHECK(bytes_read >= 0);
buffer[bytes_read] = '\0';
if (! strncmp(buffer, MSG_STOP, strlen(MSG_STOP)))
{
must_stop = 1;
}
else
{
printf("Received: %s\n", buffer);
}
} while (!must_stop);
/* cleanup */
CHECK((mqd_t)-1 != mq_close(mq));
CHECK((mqd_t)-1 != mq_unlink(QUEUE_NAME));
return 0;
}
客户端的代码：
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <mqueue.h>
#include "common.h"
int main(int argc, char **argv)
{
mqd_t mq;
char buffer[MAX_SIZE];
/* open the mail queue */
mq = mq_open(QUEUE_NAME, O_WRONLY);
CHECK((mqd_t)-1 != mq);
printf("Send to server (enter \"exit\" to stop it):\n");
do {
printf("> ");
fflush(stdout);
memset(buffer, 0, MAX_SIZE);
fgets(buffer, MAX_SIZE, stdin);
/* send the message */
CHECK(0 <= mq_send(mq, buffer, MAX_SIZE, 0));
} while (strncmp(buffer, MSG_STOP, strlen(MSG_STOP)));
/* cleanup */
CHECK((mqd_t)-1 != mq_close(mq));
return 0;
}
common 标头：
#ifndef COMMON_H_
#define COMMON_H_
#define QUEUE_NAME "/test_queue"
#define MAX_SIZE 1024
#define MSG_STOP "exit"
#define CHECK(x) \
do { \
if (!(x)) { \
fprintf(stderr, "%s:%d: ", __func__, __LINE__); \
perror(#x); \
exit(-1); \
} \
} while (0) \
#endif /* #ifndef COMMON_H_ */
编译：
gcc -o server server.c -lrt
gcc -o client client.c -lrt
【讨论】：
-
一句话。您的客户端代码缺少以下内容以使其编译：
#include <stdio.h> #include <string.h> #include <stdlib.h> -
亲爱的，我喜欢你的 CHECK 宏。
-
我确定我的理解不正确，但消息队列不应该是异步的吗？如果服务器不可用，为什么客户端会出错并退出？就我（可能是错误的）理解而言，消息队列的全部意义在于允许客户端写入无人值守的队列——否则，mqueue 和 FIFO 之间的真正区别是什么？我在这里有什么误解？你注意到我问了很多问题吗？
-
@Gutza 在这种情况下，我只想用生产者/消费者替换客户端/服务器。队列始终可通过 API 获得，操作系统将保持其安全，直到有人使用该数据。
-
@clarete，好吧，我使用的是 the_void 的术语；此外，虽然您的断言在一般情况下是正确的，但 the_void 的代码不允许客户端/生产者写入无人看管的队列（即使库允许这样做）。经过进一步考虑，我最终得到的答案是，出于某种原因，the_void“需要”在这个特定的实现中是这种情况：他/她可以选择将数据推送到队列中，而不管是否有消费者活动另一端与否，但他/她只是选择不这样做。
#include <stdio.h>
#include <fcntl.h>
#include <mqueue.h>
int main(int argc, char *argv[])
{
mqd_t mq; // message queue
struct mq_attr ma; // message queue attributes
int status = 0;
int a = 5;
int b = 0;
printf("a = %d, b = %d\n", a, b);
// Specify message queue attributes.
ma.mq_flags = 0; // blocking read/write
ma.mq_maxmsg = 16; // maximum number of messages allowed in queue
ma.mq_msgsize = sizeof(int); // messages are contents of an int
ma.mq_curmsgs = 0; // number of messages currently in queue
// Create the message queue with some default settings.
mq = mq_open("/test_queue", O_RDWR | O_CREAT, 0700, &ma);
// -1 indicates an error.
if (mq == -1)
{
printf("Failed to create queue.\n");
status = 1;
}
if (status == 0)
{
status = mq_send(mq, (char *)(&a), sizeof(int), 1);
}
if (status == 0)
{
status = mq_receive(mq, (char *)(&b), sizeof(int), NULL);
}
if ((status == 0) && (mq_close(mq) == -1))
{
printf("Error closing message queue.\n");
status = 1;
}
if ((status == 0) && (mq_unlink("test_queue") == -1))
{
printf("Error deleting message queue.\n");
status = 1;
}
printf("a = %d, b = %d\n", a, b);
return status;
}
【讨论】：
-
您的实现有一些非常错误的地方。通过 mqueue 传递指针是一个糟糕的想法，因为指针仅在其自己的进程中有效，而 mqueue 旨在用于进程之间。但最后你传递的是整数。它可能只是因为 sizeof(void*) > sizeof(int) 在大多数架构上才起作用。
-
@Juliano：谢谢，我在应该是 sizeof(int) 的地方使用 sizeof(void *)。这只是一个展示 mqueue 用法的综合示例。它演示了一个整数的内容通过队列移动到另一个整数，其中两者都被视为缓冲区。
-
@Armardeep: sizeof(a) 和 sizeof(b) 会比 sizeof(int) 好。
-
@camh：同意。我还认为，一种更好的方法（我将在生产设计中使用）是定义消息类型及其大小。任何要传输的东西都将具有加载/存储缓冲区的受控方法，并在消息通过后强制其有效性。
-
mq_open会失败，因为名字不是以/开头的，所以应该是"/test_queue"
mq_send(mq, (char *)(&a), sizeof(int), 1)从缓冲区&a复制sizeof(int)字节，在这种情况下，它不携带变量a的指针，而是携带变量a的值从一个进程到另一个进程。实施是对的。
【讨论】：
以下代码供您参考：
IPC_msgq_rcv.c
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <stdlib.h>
#define MAXSIZE 128
void die(char *s)
{
perror(s);
exit(1);
}
struct msgbuf
{
long mtype;
char mtext[MAXSIZE];
};
void main()
{
int msqid;
key_t key;
struct msgbuf rcvbuffer;
key = 1234;
if ((msqid = msgget(key, 0666)) < 0)
die("msgget()");
//Receive an answer of message type 1.
if (msgrcv(msqid, &rcvbuffer, MAXSIZE, 1, 0) < 0)
die("msgrcv");
printf("%s\n", rcvbuffer.mtext);
exit(0);
}
IPC_msgq_send.c
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#define MAXSIZE 128
void die(char *s)
{
perror(s);
exit(1);
}
struct msgbuf
{
long mtype;
char mtext[MAXSIZE];
};
main()
{
int msqid;
int msgflg = IPC_CREAT | 0666;
key_t key;
struct msgbuf sbuf;
size_t buflen;
key = 1234;
if ((msqid = msgget(key, msgflg )) < 0) //Get the message queue ID for the given key
die("msgget");
//Message Type
sbuf.mtype = 1;
printf("Enter a message to add to message queue : ");
scanf("%[^\n]",sbuf.mtext);
getchar();
buflen = strlen(sbuf.mtext) + 1 ;
if (msgsnd(msqid, &sbuf, buflen, IPC_NOWAIT) < 0)
{
printf ("%d, %ld, %s, %d \n", msqid, sbuf.mtype, sbuf.mtext, (int)buflen);
die("msgsnd");
}
else
printf("Message Sent\n");
exit(0);
}
编译每个源文件，以获得编写器可执行文件和读取器可执行文件。如下：：
gcc -o MQsender IPC_msgq_send.c
gcc -o MQreceiver IPC_msgq_rcv.c
执行每个二进制文件，您可以发送消息并从消息队列中读取消息。另外，尝试通过运行命令（在不同的队列状态）查看消息队列状态：
ipcs -q
对于您的 linux 系统，您可以通过以下方式了解 IPC 机制和可用队列等的所有详细信息：
ipcs -a
【讨论】：