Linux C语言消息队列的优先级设置

消息队列基础概念

在Linux系统中，消息队列是进程间通信（IPC, Inter - Process Communication）的一种方式。它允许不同的进程通过向队列中发送和接收消息来进行数据交换。消息队列以一种异步的方式工作，发送进程可以随时将消息添加到队列中，而接收进程可以在合适的时间从队列中取出消息。

消息队列的每个消息都由一个消息类型和消息数据组成。消息类型是一个正整数，它用于区分不同种类的消息。这为消息队列在处理不同优先级或不同用途的消息时提供了基础。

Linux系统下消息队列相关函数

msgget函数
- 功能：创建一个新的消息队列或获取一个已有的消息队列标识符。
- 函数原型：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgget(key_t key, int msgflg);

- 参数说明：
  - `key`：是一个键值，它用于唯一标识一个消息队列。通常使用`ftok`函数生成一个`key`值。
  - `msgflg`：是一组标志位，它可以指定消息队列的创建模式（如`IPC_CREAT`表示如果队列不存在则创建，`IPC_EXCL`与`IPC_CREAT`一起使用表示如果队列已存在则出错），以及权限位（类似于文件权限，如`0666`表示读写权限）。
- 返回值：成功时返回消息队列标识符，失败时返回 - 1。

2. msgsnd函数 - 功能：向指定的消息队列发送一条消息。 - 函数原型：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

- 参数说明：
  - `msqid`：是由`msgget`函数返回的消息队列标识符。
  - `msgp`：是一个指向消息结构的指针。消息结构必须以一个长整型成员开始，用于指定消息类型，后面跟着实际的消息数据。
  - `msgsz`：是消息数据部分的长度（不包括消息类型的长整型部分）。
  - `msgflg`：是一组标志位。如果设置了`IPC_NOWAIT`，当消息队列满时，`msgsnd`函数将不会阻塞，而是立即返回 - 1并设置`errno`为`EAGAIN`；否则，函数将阻塞直到有足够的空间将消息放入队列。
- 返回值：成功时返回0，失败时返回 - 1。

3. msgrcv函数 - 功能：从指定的消息队列接收一条消息。 - 函数原型：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

- 参数说明：
  - `msqid`：消息队列标识符。
  - `msgp`：指向用于存储接收消息的结构的指针。
  - `msgsz`：指定接收消息数据部分的最大长度（不包括消息类型的长整型部分）。
  - `msgtyp`：指定要接收的消息类型。如果`msgtyp`为0，则接收队列中的第一条消息；如果`msgtyp`大于0，则接收类型等于`msgtyp`的第一条消息；如果`msgtyp`小于0，则接收类型小于或等于`msgtyp`绝对值的消息中类型最小的第一条消息。
  - `msgflg`：标志位。如果设置了`IPC_NOWAIT`，当没有符合条件的消息时，`msgrcv`函数将不会阻塞，而是立即返回 - 1并设置`errno`为`ENOMSG`；如果设置了`MSG_NOERROR`，当消息数据部分长度大于`msgsz`时，将截断消息而不是返回错误。
- 返回值：成功时返回实际接收到的消息数据部分的长度，失败时返回 - 1。

4. msgctl函数 - 功能：对消息队列执行各种控制操作，如删除队列、获取或设置队列属性等。 - 函数原型：

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
int msgctl(int msqid, int cmd, struct msqid_ds *buf);

- 参数说明：
  - `msqid`：消息队列标识符。
  - `cmd`：指定要执行的控制操作。常见的操作有`IPC_STAT`（获取队列状态信息并存储到`buf`中），`IPC_SET`（根据`buf`中的信息设置队列状态），`IPC_RMID`（删除消息队列）。
  - `buf`：是一个指向`msqid_ds`结构的指针，用于传递或接收队列的相关信息。
- 返回值：成功时返回0，失败时返回 - 1。

消息队列优先级设置原理

在Linux C语言的消息队列中，消息的优先级是通过消息类型来体现的。当使用msgrcv函数接收消息时，可以通过设置msgtyp参数来指定接收特定优先级（消息类型）的消息。

例如，如果我们将高优先级的消息设置为较小的消息类型值，低优先级的消息设置为较大的消息类型值。当使用msgrcv函数并将msgtyp设置为一个较小的值时，就可以优先接收高优先级的消息。如果msgtyp设置为0，则会按照消息在队列中的顺序接收，不考虑优先级。

代码示例 - 简单的优先级消息队列

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

#define MAX_TEXT 512

// 定义消息结构
typedef struct msg_st {
    long int msg_type;
    char text[MAX_TEXT];
} message;

int main() {
    int running = 1;
    int msgid;
    key_t key;
    message data;
    size_t size = sizeof(data.text);

    // 生成一个唯一的键值
    key = ftok(".", 'a');
    if (key == -1) {
        perror("ftok");
        return 1;
    }

    // 创建消息队列
    msgid = msgget(key, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget");
        return 1;
    }

    // 发送高优先级消息
    data.msg_type = 1;
    strcpy(data.text, "High priority message");
    if (msgsnd(msgid, (void *)&data, size, 0) == -1) {
        perror("msgsnd");
        return 1;
    }

    // 发送低优先级消息
    data.msg_type = 2;
    strcpy(data.text, "Low priority message");
    if (msgsnd(msgid, (void *)&data, size, 0) == -1) {
        perror("msgsnd");
        return 1;
    }

    // 接收高优先级消息
    if (msgrcv(msgid, (void *)&data, size, 1, 0) == -1) {
        perror("msgrcv");
        return 1;
    }
    printf("Received high priority message: %s\n", data.text);

    // 接收低优先级消息
    if (msgrcv(msgid, (void *)&data, size, 2, 0) == -1) {
        perror("msgrcv");
        return 1;
    }
    printf("Received low priority message: %s\n", data.text);

    // 删除消息队列
    if (msgctl(msgid, IPC_RMID, 0) == -1) {
        perror("msgctl");
        return 1;
    }

    return 0;
}

在上述代码中：

首先使用ftok函数生成一个唯一的key值，然后通过msgget函数创建一个消息队列。
接着发送了两条消息，一条高优先级（消息类型为1），一条低优先级（消息类型为2）。
之后通过msgrcv函数，按照消息类型优先接收高优先级消息，再接收低优先级消息。
最后使用msgctl函数删除消息队列。

复杂场景下的优先级消息队列处理

在实际应用中，消息队列的优先级处理可能会更加复杂。例如，可能会有多个进程同时向消息队列发送不同优先级的消息，并且接收进程需要根据实际情况动态调整接收的优先级策略。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <unistd.h>
#include <pthread.h>

#define MAX_TEXT 512

// 定义消息结构
typedef struct msg_st {
    long int msg_type;
    char text[MAX_TEXT];
} message;

// 全局变量
int msgid;
key_t key;

// 发送消息线程函数
void *sender(void *arg) {
    int priority = *((int *)arg);
    message data;
    size_t size = sizeof(data.text);
    char msg[100];
    sprintf(msg, "Message from sender with priority %d", priority);
    data.msg_type = priority;
    strcpy(data.text, msg);

    if (msgsnd(msgid, (void *)&data, size, 0) == -1) {
        perror("msgsnd");
        pthread_exit(NULL);
    }
    pthread_exit(NULL);
}

// 接收消息线程函数
void *receiver(void *arg) {
    message data;
    size_t size = sizeof(data.text);
    long int priority = *((long int *)arg);

    if (msgrcv(msgid, (void *)&data, size, priority, 0) == -1) {
        perror("msgrcv");
        pthread_exit(NULL);
    }
    printf("Received message: %s\n", data.text);
    pthread_exit(NULL);
}

int main() {
    key = ftok(".", 'a');
    if (key == -1) {
        perror("ftok");
        return 1;
    }

    msgid = msgget(key, 0666 | IPC_CREAT);
    if (msgid == -1) {
        perror("msgget");
        return 1;
    }

    // 创建发送线程
    int priorities[] = {1, 2, 3};
    pthread_t senders[3];
    for (int i = 0; i < 3; i++) {
        if (pthread_create(&senders[i], NULL, sender, &priorities[i]) != 0) {
            perror("pthread_create");
            return 1;
        }
    }

    // 创建接收线程
    long int receive_priorities[] = {1, 2, 3};
    pthread_t receivers[3];
    for (int i = 0; i < 3; i++) {
        if (pthread_create(&receivers[i], NULL, receiver, &receive_priorities[i]) != 0) {
            perror("pthread_create");
            return 1;
        }
    }

    // 等待发送线程结束
    for (int i = 0; i < 3; i++) {
        if (pthread_join(senders[i], NULL) != 0) {
            perror("pthread_join");
            return 1;
        }
    }

    // 等待接收线程结束
    for (int i = 0; i < 3; i++) {
        if (pthread_join(receivers[i], NULL) != 0) {
            perror("pthread_join");
            return 1;
        }
    }

    // 删除消息队列
    if (msgctl(msgid, IPC_RMID, 0) == -1) {
        perror("msgctl");
        return 1;
    }

    return 0;
}

在这个代码示例中：

使用多线程模拟多个发送进程向消息队列发送不同优先级的消息。每个发送线程根据传入的优先级值设置消息类型并发送消息。
同时使用多线程模拟接收进程，每个接收线程根据传入的优先级值接收相应优先级的消息。
程序通过pthread_create创建线程，pthread_join等待线程结束，最后使用msgctl删除消息队列。

消息队列优先级设置的注意事项

消息类型的取值范围：消息类型是一个长整型，理论上取值范围很大，但实际应用中应该根据具体需求合理规划，避免使用过大或过小的值导致难以管理或与系统默认值冲突。
消息队列满的情况：当消息队列满时，如果在msgsnd函数中没有设置IPC_NOWAIT标志，发送进程将会阻塞。在高优先级消息发送频繁的情况下，可能会导致低优先级消息长时间无法发送，从而出现饥饿现象。可以通过调整消息队列的大小（通过msgctl函数的IPC_SET操作设置msg_qbytes字段）或合理设置发送标志来缓解这种情况。
接收消息的阻塞与非阻塞模式：在msgrcv函数中，IPC_NOWAIT标志决定了接收操作是否阻塞。在处理优先级消息队列时，如果设置为非阻塞模式，当没有符合优先级的消息时，函数会立即返回 - 1。这需要接收进程有合适的处理逻辑，如重试接收或进行其他任务，以避免错过重要消息。
多个进程操作消息队列：当多个进程同时对消息队列进行操作时，需要注意同步问题。例如，可以使用信号量来保证在同一时间只有一个进程对消息队列进行关键操作，防止数据竞争和不一致。

优先级消息队列在实际项目中的应用场景

实时系统：在实时系统中，某些事件需要立即处理，而有些事件可以稍后处理。通过消息队列的优先级设置，可以确保高优先级的实时事件消息优先被处理，例如传感器数据采集、紧急故障报警等。
任务调度系统：在任务调度系统中，不同的任务可能有不同的优先级。任务可以以消息的形式发送到消息队列，调度器根据消息的优先级来安排任务的执行顺序，确保重要任务优先执行。
分布式系统：在分布式系统中，不同节点之间通过消息队列进行通信。某些消息可能与系统的关键业务逻辑相关，需要优先处理，而一些普通的状态更新消息可以稍后处理。通过设置消息队列的优先级，可以优化系统的整体性能和响应速度。

通过合理设置Linux C语言消息队列的优先级，可以在不同的应用场景中提高系统的性能和响应能力，确保重要的消息能够及时得到处理，同时合理安排其他消息的处理顺序。在实际应用中，需要充分考虑系统的需求、资源限制以及可能出现的并发问题，以实现高效、稳定的消息队列优先级管理。