進程間通信之深入消息隊列的詳解

2020-01-26 16:07:20

字體：大中小

來源：轉載

供稿：網(wǎng)友

最近在Hi3515上調試Qt與DVR程序，發(fā)現(xiàn)他們之間使用消息隊列通信的，閑暇之余，就總結了一下消息隊列，呵呵，自認為通俗易懂，同時，在應用中也發(fā)現(xiàn)了消息隊列的強大之處。

關于線程的管理(互斥量和條件變量)見：Linux線程管理必備:解析互斥量與條件變量的詳解

一、消息隊列的特點

  1.消息隊列是消息的鏈表,具有特定的格式,存放在內存中并由消息隊列標識符標識.
  2.消息隊列允許一個或多個進程向它寫入與讀取消息.
  3.管道和命名管道都是通信數(shù)據(jù)都是先進先出的原則。
  4.消息隊列可以實現(xiàn)消息的隨機查詢,消息不一定要以先進先出的次序讀取,也可以按消息的類型讀取.比FIFO更有優(yōu)勢。

目前主要有兩種類型的消息隊列：POSIX消息隊列以及系統(tǒng)V消息隊列，系統(tǒng)V消息隊列目前被大量使用。系統(tǒng)V消息隊列是隨內核持續(xù)的，只有在內核重起或者人工刪除時，該消息隊列才會被刪除。

二、相關函數(shù)

1. 獲得key值

key_t ftok(char *pathname, int projid)

#include <sys/types.h>
#include <sys/ipc.h>
參數(shù):
pathname：文件名（含路徑）,通常設置為當前目錄“.” 比如projid為'a',則為"./a"文件
projid：項目ID，必須為非0整數(shù)（0-255）.

2. 創(chuàng)建消息隊列
int msgget(key_t key, int msgflag)
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
功能：
用于創(chuàng)建一個新的或打開一個已經(jīng)存在的消息隊列，此消息隊列與key相對應。
參數(shù)：
  key：函數(shù)ftok的返回值或IPC_PRIVATE。
  msgflag：
      IPC_CREAT:創(chuàng)建新的消息隊列。
      IPC_EXCL:與IPC_CREAT一同使用，表示如果要創(chuàng)建的消息隊列已經(jīng)存在，則返回錯誤。
      IPC_NOWAIT:讀寫消息隊列要求無法滿足時，不阻塞。
返回值：
  調用成功返回隊列標識符,否則返回-1.

在以下兩種情況下，將創(chuàng)建一個新的消息隊列：
1、如果沒有與鍵值key相對應的消息隊列，并且msgflag中包含了IPC_CREAT標志位。
2、key參數(shù)為IPC_PRIVATE。

3. 消息隊列屬性控制
int msgctl(int msqid, int cmd, struct msqid_ds *buf)
功能:
對消息隊列進行各種控制操作,操作的動作由cmd控制。
參數(shù):
  msqid:消息隊列ID，消息隊列標識符，該值為msgget創(chuàng)建消息隊列的返回值。
  cmd:
      IPC_STAT:將msqid相關的數(shù)據(jù)結構中各個元素的當前值存入到由buf指向的結構中.
      IPC_SET:將msqid相關的數(shù)據(jù)結構中的元素設置為由buf指向的結構中的對應值.
      IPC_RMID:刪除由msqid指示的消息隊列,將它從系統(tǒng)中刪除并破壞相關數(shù)據(jù)結構.
buf:消息隊列緩沖區(qū)
     struct msqid_ds {
               struct ipc_perm msg_perm; /* Ownership and permissions*/
               time_t         msg_stime; /* Time of last msgsnd() */
               time_t         msg_rtime; /* Time of last msgrcv() */
               time_t         msg_ctime; /* Time of last change */
               unsigned long __msg_cbytes; /* Current number of bytes in queue (non-standard) */
               msgqnum_t      msg_qnum; /* Current number of messages in queue */
               msglen_t       msg_qbytes; /* Maximum number of bytesallowed in queue */
               pid_t          msg_lspid; /* PID of last msgsnd() */
               pid_t          msg_lrpid; /* PID of last msgrcv() */
                     };

     struct ipc_perm {
               key_t key;                        /* Key supplied to msgget() */
               uid_t uid; /* Effective UID of owner */
               gid_t gid; /* Effective GID of owner */
               uid_t cuid; /* Effective UID of creator */
               gid_t cgid; /* Effective GID of creator */
               unsigned short mode;    /* Permissions */
               unsigned short seq; /* Sequence number */
                   };

4．發(fā)送信息到消息隊列
int msgsnd(int msqid, struct msgbuf *msgp, size_t msgsz, int msgflag)
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
功能:

將新消息添加到隊列尾端,即向消息隊列中發(fā)送一條消息。
參數(shù)：
  msqid:已打開的消息隊列id
  msgp:存放消息的結構體指針。
  msgflag:函數(shù)的控制屬性。
  消息結構msgbuf為:
  struct msgbuf
  {
         long mtype;//消息類型
         char mtext[1];//消息正文，消息數(shù)據(jù)的首地址，這個數(shù)據(jù)的最大長度為8012吧，又可把他看成是一個結構，也有類型和數(shù)據(jù)，recv時解析即可。
  }
  msgsz:消息數(shù)據(jù)的長度。
  msgflag:
       IPC_NOWAIT: 指明在消息隊列沒有足夠空間容納要發(fā)送的消息時，msgsnd立即返回。
       0:msgsnd調用阻塞直到條件滿足為止.（一般選這個）

5. 從消息隊列接收信息
ssize_t msgrcv(int msqid, struct msgbuf *msgp, size_t msgsz, long msgtype, int msgflag)
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
功能:

從隊列中接收消息
參數(shù):
  msqid:已打開的消息隊列id
  msgp:存放消息的結構體指針。msgp->mtype與第四個參數(shù)是相同的。
  msgsz:消息的字節(jié)數(shù)，指定mtext的大小。
  msgtype:消息類型，消息類型 mtype的值。如果為0，則接受該隊列中的第一條信息，如果小于0，則接受小于該值的絕對值的消息類型，如果大于0，接受指定類型的消息，即該值消息。
  msgflag:函數(shù)的控制屬性。
  msgflag:
      MSG_NOERROR:若返回的消息比nbytes字節(jié)多,則消息就會截短到nbytes字節(jié),且不通知消息發(fā)送進程.
      IPC_NOWAIT:調用進程會立即返回.若沒有收到消息則返回-1.
      0:msgrcv調用阻塞直到條件滿足為止.
在成功地讀取了一條消息以后，隊列中的這條消息將被刪除。

三、相關例子

例1：消息隊列之簡單收發(fā)測試
#include <sys/types.h>
#include <sys/msg.h>
#include <unistd.h>
#include <string.h>
#include <stdio.h>

struct msg_buf{
int mtype;//消息類型
char data[255];//數(shù)據(jù)
};

int main(int argc, char *argv[])
{
    key_t key;
    int msgid;
    int ret;
    struct msg_buf msgbuf;
      //獲取key值
    key = ftok(".", 'a');
      printf("key = [%x]/n", key);

    //創(chuàng)建消息隊列
    msgid = msgget(key, IPC_CREAT|0666);/*通過文件對應*/
      if(msgid == -1)
      {
            printf("creat error/n");
            return -1;
    }

      //以當前進程類型，非阻塞方式發(fā)送"test data"到消息隊列
    msgbuf.mtype = getpid();
      strcpy(msgbuf.data, "test data");
      ret = msgsnd(msgid, &msgbuf, sizeof(msgbuf.data), IPC_NOWAIT);
    if(ret == -1)
    {
            printf("send message err/n");
          return -1;
    }

      //以非阻塞方式接收數(shù)據(jù)
    memset(&msgbuf, 0, sizeof(msgbuf));
    ret = msgrcv(msgid, &msgbuf, sizeof(msgbuf.data), getpid(), IPC_NOWAIT);
    if(ret == -1)
    {
          printf("receive message err/n");
            return -1;
    }
    printf("receive msg = [%s]/n", msgbuf.data);
    return 0;
}
例2：進程間消息隊列通信
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <string.h>
#include <signal.h>

struct msgbuf{
int mtype;
char mtext[100];
};

int main(void)
{
    key_t key;
    pid_t pid;
      int msgid;
      struct msgbuf msg;

      key=ftok(".", 0x01);
      if ( (msgid = msgget(key, IPC_CREAT|0666)) <0 )
      {
            perror("msgget error");
            exit(1);
    }

      //創(chuàng)建一個進程
    if ( (pid = fork()) < 0 )
    {
          perror("fork error");
          exit(1);
    }
      //子進程收信息
    else if (pid==0)
    {
            while(1)
          {
                  memset(msg.mtext, 0, 100);
                msgrcv(msgid, &msg, 100, 2, 0); //receive the msg from 2
                  printf("/receive:%s/n:", msg.mtext);
                  fflush(stdout);
            }
            exit(0);
      }
    //父進程發(fā)信息
    else
    {
            while(1)
          {
                  memset(msg.mtext, 0, 100);
                  printf("father:");
                  fgets(msg.mtext, 100, stdin);
                  if (strncmp("bye", msg.mtext, 3)==0)//如果前3個字符為bye,則退出
                  {
                        kill(pid, SIGSTOP);
                        exit(1);
                  }
                  msg.mtype=1;//send to 1
                  msg.mtext[strlen(msg.mtext)-1]='/0';
                  msgsnd(msgid, &msg, strlen(msg.mtext)+1, 0);
            }
    }