C语言消息队列设计（从零开始实现高效进程间通信）

什么是消息队列？

消息队列是一种先进先出（FIFO）的数据结构，用于在生产者和消费者之间传递消息。在多任务环境中，它可以解耦发送方和接收方，提高系统的稳定性和可维护性。尤其在资源受限的嵌入式系统消息队列应用中，自定义轻量级队列比使用系统IPC更灵活。

设计目标

• 支持任意类型的消息（通过 void* 指针）
• 线程安全（使用互斥锁）
• 固定容量，防止内存无限增长
• 提供阻塞/非阻塞发送与接收接口

数据结构定义

我们首先定义消息队列的核心结构：

#include <stdio.h>#include <stdlib.h>#include <pthread.h>#include <string.h>typedef struct {    void* data;        // 消息内容指针    size_t size;       // 消息大小（字节）} Message;typedef struct {    Message* buffer;   // 消息缓冲区数组    int capacity;      // 队列最大容量    int head;          // 出队索引    int tail;          // 入队索引    int count;         // 当前消息数量    pthread_mutex_t mutex;     // 互斥锁    pthread_cond_t not_full;   // 非满条件变量    pthread_cond_t not_empty;  // 非空条件变量} MsgQueue;

初始化与销毁

创建队列并分配内存：

// 初始化消息队列MsgQueue* msg_queue_create(int capacity) {    if (capacity <= 0) return NULL;    MsgQueue* queue = (MsgQueue*)malloc(sizeof(MsgQueue));    if (!queue) return NULL;    queue->buffer = (Message*)calloc(capacity, sizeof(Message));    if (!queue->buffer) {        free(queue);        return NULL;    }    queue->capacity = capacity;    queue->head = 0;    queue->tail = 0;    queue->count = 0;    pthread_mutex_init(&queue->mutex, NULL);    pthread_cond_init(&queue->not_full, NULL);    pthread_cond_init(&queue->not_empty, NULL);    return queue;}// 销毁消息队列void msg_queue_destroy(MsgQueue* queue) {    if (!queue) return;    // 释放每条消息的数据    for (int i = 0; i < queue->capacity; i++) {        if (queue->buffer[i].data) {            free(queue->buffer[i].data);        }    }    free(queue->buffer);    pthread_mutex_destroy(&queue->mutex);    pthread_cond_destroy(&queue->not_full);    pthread_cond_destroy(&queue->not_empty);    free(queue);}

发送与接收消息

以下是核心的入队（send）和出队（receive）函数，支持阻塞模式：

// 阻塞式发送消息int msg_queue_send(MsgQueue* queue, const void* data, size_t size) {    if (!queue || !data || size == 0) return -1;    pthread_mutex_lock(&queue->mutex);    // 如果队列满，等待 not_full 信号    while (queue->count == queue->capacity) {        pthread_cond_wait(&queue->not_full, &queue->mutex);    }    // 复制消息数据    void* copy = malloc(size);    if (!copy) {        pthread_mutex_unlock(&queue->mutex);        return -1;    }    memcpy(copy, data, size);    queue->buffer[queue->tail].data = copy;    queue->buffer[queue->tail].size = size;    queue->tail = (queue->tail + 1) % queue->capacity;    queue->count++;    // 唤醒等待接收的线程    pthread_cond_signal(&queue->not_empty);    pthread_mutex_unlock(&queue->mutex);    return 0;}// 阻塞式接收消息int msg_queue_receive(MsgQueue* queue, void** out_data, size_t* out_size) {    if (!queue || !out_data || !out_size) return -1;    pthread_mutex_lock(&queue->mutex);    // 如果队列空，等待 not_empty 信号    while (queue->count == 0) {        pthread_cond_wait(&queue->not_empty, &queue->mutex);    }    Message* msg = &queue->buffer[queue->head];    *out_data = msg->data;    *out_size = msg->size;    // 注意：调用者需负责 free(*out_data)    msg->data = NULL; // 避免重复释放    msg->size = 0;    queue->head = (queue->head + 1) % queue->capacity;    queue->count--;    // 唤醒等待发送的线程    pthread_cond_signal(&queue->not_full);    pthread_mutex_unlock(&queue->mutex);    return 0;}

使用示例

下面是一个简单的生产者-消费者模型：

void* producer(void* arg) {    MsgQueue* q = (MsgQueue*)arg;    for (int i = 0; i < 5; i++) {        char msg[32];        snprintf(msg, sizeof(msg), "Message %d", i);        msg_queue_send(q, msg, strlen(msg) + 1);        printf("Sent: %s\n", msg);        usleep(100000); // 模拟工作    }    return NULL;}void* consumer(void* arg) {    MsgQueue* q = (MsgQueue*)arg;    for (int i = 0; i < 5; i++) {        void* data;        size_t size;        msg_queue_receive(q, &data, &size);        printf("Received: %s\n", (char*)data);        free(data); // 释放消息内存    }    return NULL;}int main() {    MsgQueue* q = msg_queue_create(3); // 容量为3    pthread_t prod, cons;    pthread_create(&prod, NULL, producer, q);    pthread_create(&cons, NULL, consumer, q);    pthread_join(prod, NULL);    pthread_join(cons, NULL);    msg_queue_destroy(q);    return 0;}

总结

通过本教程，你已经掌握了如何从零开始实现一个线程安全的消息队列实现教程。这个设计适用于大多数需要可靠通信的场景，特别是在无法使用系统级消息队列（如POSIX mq）的嵌入式系统消息队列项目中。你可以根据实际需求扩展超时机制、优先级队列等功能。

记住：消息队列的核心在于“解耦”与“缓冲”。合理使用它，能让你的C语言程序更加健壮、可维护。

关键词回顾：C语言消息队列、嵌入式系统消息队列、进程间通信C语言、消息队列实现教程