当前位置：首页 > news >正文

linux多线(进)程编程——（7）消息队列

news 来源：原创 2025/4/20 22:38:41

前言

现在修真界大家的沟通手段已经越来越丰富了，有了匿名管道，命名管道，共享内存等多种方式。但是随着深入使用人们逐渐发现了这些传音术的局限性。
匿名管道：只能在有血缘关系的修真者（进程）间使用，局限性大。
命名管道：需要创建出一个文件，并通过文件方式沟通，很不方便。
共享内存：这个就更可怕了，传递速度太快，大量无关信息也可能会涌入脑海，而且两个人如果想法有冲突就会把各自的思路相互覆盖。（这个涉及到到进程间同步的一些知识，后续解决。）
无数心怀远大抱负的修士期待着能解决这些问题，从而一句成名。很快，有人成功了，他提出了一种新的沟通方式（消息队列）。

消息队列

消息队列，消息是定语，修饰队列，也就是装消息的队列，本质上是一个链表（这里涉及到队列的两种实现形式：数组环形实现和链表实现）。
为什么使用链表实现消息队列呢？因为这个队列并不是十分纯粹的先入先出操作。偶尔会涉及到消息的筛选。所以使用链表来实现。
消息队列由进程创建，但是它位于内核空间，由内核管理，用于在不同进程间传递数据。
在这里插入图片描述

创建一个消息队列 / 获取一个已经存在的消息队列id

int msgget(key_t key, int msgflg);

其中各个参数的含义是：
key：键值，操作系统中唯一
msgflg：标志位，用于设置用户权限
返回值：消息队列的id，用来表示这个消息队列。
当消息队列已经存在时，我们会字节根据键值索引得到这个消息队列的id。之后我们操作消息队列都是使用这个id而不是键值。

这里插一句，大家可以发现，在linux中经常有这种一个id代表一个对象的管理手段，无论是文件，还是共享内存，还是今天学的消息队列都是这样。未来这种手段我们还会见到更多，大家要学会把他们联系起来。

先创建一个消息队列：

#include <stdio.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>int main() {int id = msgget(1, IPC_CREAT);return 0;
}

创建成功后我们使用ipcs指令查看系统中的消息队列（这个我指令在共享内存中讲过）。从结果中我们可以看到现在我们的系统上有一个key值为1的消息队列，他的id是0。

lol@hyl:~/work/linux_study/msgque$ ipcs------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages    
0x00000001 0          hyl        0          0            0           ------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status      ------ Semaphore Arrays --------
key        semid      owner      perms      nsems

大家可以发现，在我输入ipcs指令时我的程序应该已经停止运行了，但是我们的消息队列仍然存在，这个与之前将的共享内存类似。因此如果我们在程序中创建了消息队列而不及时释放，会引起严重的内存泄漏。这种内存泄漏更加严重，因为即使进程停止了它也不会被系统回收。
在这里我们手动删除消息队列，输入ipcrm -q 0（这个指令也是我们之前讲过的，-q代表消息对了 -m代表共享内存。）

lol@hyl:~/work/linux_study/msgque$ ipcrm -q 0
lol@hyl:~/work/linux_study/msgque$ ipcs------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages    ------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status      ------ Semaphore Arrays --------
key        semid      owner      perms      nsems

创建好了消息队列后我们如何实现进程间通信呢？
发送消息

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

参数的意义：
msqid：消息队列的id
msgp：存放消息的缓冲区，类型为struct msgbuf，定义如下：

#define DATA_SIZE	128
// 在源文件中仿造这个格式自行定义即可
struct msgbuf{long mtype;//消息类型（>0）char mtext[DATA_SIZE];//消息文本
};

msgsz：发送数据的大小，这里是msgp结构体中mtext的大小DATA_SIZE。
msgflg：是否阻塞，为0表示阻塞发送

接收消息：
在struct msgbuf类型的结构体中，mtype成员的作用主要是在接受数据时设置优先级，接收消息的函数定义为：

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp,int msgflg);

其余参数与上面发送函数相同，主要是：
msgtyp：表示接收数据的类型。
当msgtyp = 0时，则返回队头数据，暨最先进入队列的数据。
当msgtyp > 0时，返回队列中消息类型（暨msgp->mtype）等于msgtyp 的数据。
当msgtyp < 0时，返回其类型小于或等于mtype参数的绝对值的最小的一个消息。
有了这个参数我们就可以实现关键消息优先处理的机制，越关键的消息我们就将他的消息类型设置的越小，这样我们就可以使用msgtyp < 0来优先处理高优先级的消息。

关于消息队列的阻塞与非阻塞
对于接收来说，阻塞是指消息队列为空或者没有需要的消息类型是，程序停在msgrcv处等待消息的到来。
对于发送来说，阻塞是指消息队列内部容量满了之后，程序停在msgsnd处等待消息队列内部的位置空出。
以消息接收为例（进程p1）：

int main() {int id = msgget(1, 0666 | IPC_CREAT);printf( "id = : %d\n", id );struct msgbuf* buf = (struct msgbuf *)malloc(sizeof(struct msgbuf));msgrcv(id, buf, 128, 0, 0);return 0;
}

对于这段代码，我创建消息队列后直接调用msgrcv接收消息队列，由于此时消息队列为空，因此系统会被阻塞，无法顺利退出，直到其他进程向消息队列内部发送数据。

lol@hyl:~/work/linux_study/msgque$ ./p1
id = : 0# 这里没有输出，因为程序一直被阻塞

这时我们尝试使用另一个进程向该进程发送数据（进程p2）

int main() {int id = msgget(1, 0666 | IPC_CREAT);printf( "id = : %d\n", id );struct msgbuf* buf = (struct msgbuf *)malloc(sizeof(struct msgbuf));strcpy(buf->data, "hello, world!");msgsnd(id, buf, 128, 0);return 0;
}

后台运行p1，这里显示p1正在运行，pid为134070

lol@hyl:~/work/linux_study/msgque$ ./p1&
[1] 134070
id = : 0

运行p2，终端输出了[1]进程运行完毕，证明了进程1确实被阻塞，等待消息队列内的消息

lol@hyl:~/work/linux_study/msgque$ ./p1&
[1] 134070
id = : 0
lol@hyl:~/work/linux_study/msgque$ ./p2
id = : 0
hello, world![1]+  Done                    ./p1

当我们将p1的msgrcv更改为：msgrcv(id, buf, 128, 0, IPC_NOWAIT);

int main() {int id = msgget(1, 0666 | IPC_CREAT);printf( "id = : %d\n", id );struct msgbuf* buf = (struct msgbuf *)malloc(sizeof(struct msgbuf));msgrcv(id, buf, 128, 0, IPC_NOWAIT);printf("%s", buf->data);return 0;
}

运行程序可以看到终端直接再次进入了等待模式，证明消息队列没有阻塞。

lol@hyl:~/work/linux_study/msgque$ ./p1
id = : 0
lol@hyl:~/work/linux_study/msgque$

控制消息队列

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

这个接口的最常见的用法作用就是用于删除消息队列，当cmd = IPC_RMID时：

msgctl(msgid, IPC_RMID, NULL);//将队列从系统内核中删除

完整代码

proc1.c：进程1阻塞接收数据

#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>struct msgbuf {int type;char data[128];
};int main() {int id = msgget(1, 0666 | IPC_CREAT);printf( "id = : %d\n", id );struct msgbuf* buf = (struct msgbuf *)malloc(sizeof(struct msgbuf));msgrcv(id, buf, 128, 0, 0);     // 阻塞等待printf("%s", buf->data);        // 打印数据msgctl(id, IPC_RMID, NULL);     // 删除消息队列return 0;
}