无涯教程-进程 - 消息队列

使用消息队列的通信可以通过以下方式进行:

• 通过一个进程写入共享内存，并通过另一个进程从共享内存读取。我们知道，读取也可以通过多个进程完成。

• 由具有不同数据包的一个进程写入共享内存，并由多个进程(即根据消息类型)从共享内存中读取。

看完消息队列上的某些信息后，现在该检查支持消息队列的系统调用(System V)。

要使用消息队列执行通信，请执行以下步骤-

步骤1  -  创建消息队列或连接到已存在的消息队列(msgget())

步骤2  -  写入消息队列(msgsnd())

步骤3  -  从消息队列中读取(msgrcv())

步骤4  -  对消息队列(msgctl())执行控制操作

现在，让我们检查上述调用的语法和某些信息。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgget(key_t key, int msgflg)

该系统调用创建或分配系统V消息队列。以下参数需要传递-

• 第一个参数key:识别消息队列，密钥可以是任意值，也可以是从库函数ftok()派生的值。

• 第二个参数shmflg:指定所需的消息队列标志，例如IPC_CREAT(如果不存在则创建消息队列)或IPC_EXCL(与IPC_CREAT一起使用以创建消息队列，如果消息失败，则调用失败)。还需要传递权限。

该调用将在成功时返回有效的消息队列标识符，在失败的情况下返回-1。要知道失败的原因，请使用errno变量或perror()函数进行检查。

与此相关的各种错误包括EACCESS(拒绝权限)，EEXIST(无法创建已经存在的队列)，ENOENT(没有队列)，ENOMEM(没有足够的内存来创建队列)等。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgsnd(int msgid, const void *msgp, size_t msgsz, int msgflg)

此系统调用将消息发送/追加到消息队列(系统V)。以下参数需要传递-

• 第一个参数msgid识别消息队列，即消息队列标识符， msgget()成功后将接收到标识符值

• 第二个参数msgp是指向发送到调用方的消息的指针，该消息的格式如下-

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

变量mtype用于与不同的消息类型进行通信，这在msgrcv()调用中进行了详细说明，变量mtext是一个数组或其他结构，其大小由msgsz(正值)指定。如果未提及多行文本字段，则将其视为零大小消息，这是允许的。

• 第三个参数msgsz是消息的大小(消息应以空字符结尾)

• 第四个参数msgflg表示某些标志，例如IPC_NOWAIT(在队列中未找到消息时立即返回或MSG_NOERROR立即返回)

成功时此调用将返回0，失败则返回-1。要知道失败的原因，请使用errno变量或perror()函数进行检查。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgrcv(int msgid, const void *msgp, size_t msgsz, long msgtype, int msgflg)

该系统调用从消息队列(系统V)中检索消息。以下参数需要传递-

• 第一个参数msgid识别消息队列，即消息队列标识符， msgget()成功后将接收到标识符值

• 第二个参数msgp是从调用方收到的消息的指针。它以以下形式的结构定义-

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

变量mtype用于与不同的消息类型进行通信，变量mtext是一个数组或其他结构，其大小由msgsz指定。如果未提及多行文本字段，则将其视为零大小消息，这是允许的。

• 第三个参数msgsz是接收到的消息的大小(消息应以空字符结尾)

• 第四个参数msgtype指示消息的类型-

  • 如果msgtype为0        -  读取队列中第一个收到的消息

  • 如果msgtype为+ve   -   读取类型为msgtype的队列中的第一条消息

  • 如果msgtype为-ve    -    读取最小类型小于或等于消息类型绝对值的第一条消息

• 第五个参数msgflg表示某些标志，例如IPC_NOWAIT(当在队列中未找到消息时立即返回或MSG_NOERROR(如果超过msgsz字节则截断消息文本)

此调用将在成功时返回mtext数组中实际接收的字节数，如果失败则返回-1。要知道失败的原因，请使用errno变量或perror()函数进行检查。

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

int msgctl(int msgid, int cmd, struct msqid_ds *buf)

该系统调用执行消息队列(系统V)的控制操作。以下参数需要传递-

• 第一个参数msgid识别消息队列，即消息队列标识符， msgget()成功后将接收到标识符值

• 第二个参数cmd是用于在消息队列上执行所需控制操作的命令。 cmd的有效值为-

            IPC_STAT    -   将结构msqid_ds的每个成员的当前值的信息复制到buf指向的传递结构。此命令需要对消息队列具有读取权限。

            IPC_SET       -  设置结构buf指向的用户ID，所有者的组ID，权限等。

            IPC_RMID   -  立即删除消息队列。

            IPC_INFO   -  返回有关由buf指向的结构类型为msginfo的消息队列限制和参数的信息

            MSG_INFO-  返回一个msginfo结构，该结构包含有关消息队列消耗的系统资源的信息。

• 第三个参数buf是指向名为struct msqid_ds的消息队列结构的指针。此结构的值将用于按cmd设置或获取。

该调用将根据传递的命令返回该值。 IPC_INFO和MSG_INFO或MSG_STAT成功返回消息队列的索引或标识符，对于其他操作返回0，在失败的情况下返回-1。要知道失败的原因，请使用errno变量或perror()函数进行检查。

看完有关消息队列的基本信息和系统调用之后，现在该检查程序了。

让我们在看程序之前先看一下描述-

步骤1  -  创建两个进程，一个用于发送到消息队列(msgq_send.c)，另一个用于从消息队列中检索(msgq_recv.c)

步骤2  -  使用ftok()函数创建密钥。为此，最初将创建文件msgq.txt以获取唯一密钥。

步骤3 -  发送进程执行以下操作。

• 读取用户输入的字符串

• 删除新行(如果存在)

• 发送到消息队列

• 重复该进程直到输入(CTRL + D)结束

• 接收到输入结束后，发送消息" end"以表示进程结束

步骤4 -  在接收进程中，执行以下操作。

• 从队列中读取消息
• 显示输出
• 如果收到的消息是"end",请完成该进程并退出

为简化起见，我们不在此示例中使用消息类型。同样，一个进程正在写入队列，而另一个进程正在从队列读取。可以根据需要扩展它，即，理想情况下，一个进程将写入队列，而多个进程则从队列中读取。

现在，让我们检查进程(消息发送到队列中)–文件:msgq_send.c

/* Filename: msgq_send.c */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define PERMS 0644
struct my_msgbuf {
   long mtype;
   char mtext[200];
};

int main(void) {
   struct my_msgbuf buf;
   int msqid;
   int len;
   key_t key;
   system("touch msgq.txt");
   
   if ((key = ftok("msgq.txt", 'B')) == -1) {
      perror("ftok");
      exit(1);
   }
   
   if ((msqid = msgget(key, PERMS | IPC_CREAT)) == -1) {
      perror("msgget");
      exit(1);
   }
   printf("message queue: ready to send messages.\n");
   printf("Enter lines of text, ^D to quit:\n");
   buf.mtype = 1; /* we don't really care in this case */
   
   while(fgets(buf.mtext, sizeof buf.mtext, stdin) != NULL) {
      len = strlen(buf.mtext);
      /* remove newline at end, if it exists */
      if (buf.mtext[len-1] == '\n') buf.mtext[len-1] = '\0';
      if (msgsnd(msqid, &buf, len+1, 0) == -1) /* +1 for '\0' */
      perror("msgsnd");
   }
   strcpy(buf.mtext, "end");
   len = strlen(buf.mtext);
   if (msgsnd(msqid, &buf, len+1, 0) == -1) /* +1 for '\0' */
   perror("msgsnd");
   
   if (msgctl(msqid, IPC_RMID, NULL) == -1) {
      perror("msgctl");
      exit(1);
   }
   printf("message queue: done sending messages.\n");
   return 0;
}

编译和执行步骤

message queue: ready to send messages.
Enter lines of text, ^D to quit:
this is line 1
this is line 2
message queue: done sending messages.

以下是消息接收进程中的代码(从队列中检索消息)–文件:msgq_recv.c

/* Filename: msgq_recv.c */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define PERMS 0644
struct my_msgbuf {
   long mtype;
   char mtext[200];
};

int main(void) {
   struct my_msgbuf buf;
   int msqid;
   int toend;
   key_t key;
   
   if ((key = ftok("msgq.txt", 'B')) == -1) {
      perror("ftok");
      exit(1);
   }
   
   if ((msqid = msgget(key, PERMS)) == -1) { /* connect to the queue */
      perror("msgget");
      exit(1);
   }
   printf("message queue: ready to receive messages.\n");
   
   for(;;) { /* normally receiving never ends but just to make conclusion 
             /* this program ends wuth string of end */
      if (msgrcv(msqid, &buf, sizeof(buf.mtext), 0, 0) == -1) {
         perror("msgrcv");
         exit(1);
      }
      printf("recvd:\"%s\"\n", buf.mtext);
      toend = strcmp(buf.mtext,"end");
      if (toend == 0)
      break;
   }
   printf("message queue: done receiving messages.\n");
   system("rm msgq.txt");
   return 0;
}

编译和执行步骤

message queue: ready to receive messages.
recvd: "this is line 1"
recvd: "this is line 2"
recvd: "end"
message queue: done receiving messages.

参考链接

https://www.learnfk.com/process/inter-process-communication-message-queues.html