消息队列数据结构：数组实现环形队列、链表实现链式队列、队列的线程安全设计

消息队列这东西，说白了就是任务之间的「快递中转站」。我在嵌入式项目里用过不下十种队列实现，从裸机到RTOS，从单核到多核，踩过的坑能写满一本笔记本。今天咱们就把队列的三种核心形态掰开揉碎了讲清楚。

一、数组实现环形队列

环形队列，也叫循环缓冲区，是我个人最常用的队列实现方式。为什么？因为它简单、高效、没有动态内存分配。

核心思想：用固定大小的数组，通过头尾指针的循环移动来实现先进先出。当指针到达数组末尾时，自动折回到开头。

// 环形队列结构体定义
typedef struct {
    uint32_t *buffer;   // 数据缓冲区
    int head;           // 队头索引（出队位置）
    int tail;           // 队尾索引（入队位置）
    int size;           // 队列容量
    int count;          // 当前元素个数（可选）
} ring_queue_t;

嗯，这里要注意一个关键点：如何区分「队列满」和「队列空」？ 我刚开始做的时候，直接用 head == tail 判断空，结果队列满的时候 head 也等于 tail，直接翻车了。

常见的解决方案有两种：

  • 预留一个空位法：永远不让 tail 追上 head，即 (tail + 1) % size == head 表示满
  • 计数法：维护一个 count 变量，入队加一，出队减一

我个人偏爱计数法，虽然多占一个变量，但逻辑清晰，不容易出 bug。我在一个电机控制项目中就吃过预留空位法的亏——队列大小是 2 的幂，取模运算被优化成了位运算，结果预留空位导致实际可用空间少了一个，调试了半天才发现。

// 入队操作（计数法）
int ring_enqueue(ring_queue_t *q, uint32_t data) {
    if (q->count >= q->size) {
        return -1;  // 队列满
    }
    q->buffer[q->tail] = data;
    q->tail = (q->tail + 1) % q->size;
    q->count++;
    return 0;
}

// 出队操作
int ring_dequeue(ring_queue_t *q, uint32_t *data) {
    if (q->count <= 0) {
        return -1;  // 队列空
    }
    *data = q->buffer[q->head];
    q->head = (q->head + 1) % q->size;
    q->count--;
    return 0;
}

小技巧：如果队列大小是 2 的幂，可以用位运算代替取模：tail = (tail + 1) & (size - 1)。性能能提升不少，尤其是在中断服务函数里。

二、链表实现链式队列

链式队列的优势在于动态大小，适合消息长度不确定或者数量变化大的场景。但代价是每次入队出队都要 malloc/free，在嵌入式系统里这是把双刃剑。

// 链式队列节点
typedef struct queue_node {
    void *data;                 // 数据指针
    struct queue_node *next;    // 下一个节点
} queue_node_t;

// 链式队列控制块
typedef struct {
    queue_node_t *front;    // 队头
    queue_node_t *rear;     // 队尾
    int count;              // 节点数量
} linked_queue_t;

链式队列的入队操作很简单：新建节点，挂到队尾，更新 rear 指针。出队则是从 front 取节点，释放内存。

// 链式队列入队
int linked_enqueue(linked_queue_t *q, void *data) {
    queue_node_t *node = (queue_node_t *)malloc(sizeof(queue_node_t));
    if (!node) return -1;
    
    node->data = data;
    node->next = NULL;
    
    if (q->rear) {
        q->rear->next = node;
    } else {
        q->front = node;  // 第一个节点
    }
    q->rear = node;
    q->count++;
    return 0;
}

我曾经踩过的坑：在中断服务函数里用 malloc！中断上下文可能持有锁，malloc 内部有锁保护，直接死锁。后来我改用内存池 + 链式队列，才彻底解决这个问题。

链式队列还有一个隐藏问题——内存碎片。频繁的 malloc/free 会导致堆碎片化，运行几天后突然分配失败。我的建议是：如果消息大小固定，优先用环形队列；如果必须用链式，配合内存池使用。

三、队列的线程安全设计

这才是真正考验功力的地方。单线程环境下的队列实现再花哨，放到多线程里分分钟崩给你看。

线程安全的核心问题就三个：

  • 互斥：多个线程不能同时修改队列状态
  • 同步：生产者通知消费者「有数据了」，消费者通知生产者「有空间了」
  • 原子性：入队/出队操作不能被中断

3.1 互斥锁方案

最直接的做法：每个队列配一个 mutex，入队出队前加锁，操作完解锁。

// 线程安全环形队列
typedef struct {
    ring_queue_t ring;
    pthread_mutex_t lock;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
} safe_ring_queue_t;

int safe_enqueue(safe_ring_queue_t *q, uint32_t data) {
    pthread_mutex_lock(&q->lock);
    while (ring_is_full(&q->ring)) {
        pthread_cond_wait(&q->not_full, &q->lock);
    }
    ring_enqueue(&q->ring, data);
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->lock);
    return 0;
}

这个方案在 Linux 上跑没问题，但在裸机或 RTOS 上，pthread 那一套可能用不了。我一般在 FreeRTOS 上用队列 API 封装一层，或者直接用信号量。

3.2 无锁队列（Lock-Free）

如果追求极致性能，或者不允许阻塞（比如中断上下文），可以考虑无锁队列。核心思路是利用原子操作 CAS（Compare And Swap）来实现并发安全。

// 无锁环形队列（单生产者单消费者场景）
typedef struct {
    uint32_t buffer[QUEUE_SIZE];
    volatile int head;  // volatile 防止编译器优化
    volatile int tail;
} lockfree_queue_t;

int lockfree_enqueue(lockfree_queue_t *q, uint32_t data) {
    int tail = q->tail;
    int next = (tail + 1) % QUEUE_SIZE;
    if (next == q->head) return -1;  // 满
    
    q->buffer[tail] = data;
    __sync_synchronize();  // 内存屏障
    q->tail = next;
    return 0;
}

重要提醒：无锁队列只适合单生产者单消费者场景。多生产者多消费者需要更复杂的 CAS 循环，而且 ABA 问题、内存重排序问题非常棘手。我建议除非性能瓶颈明确在锁上，否则别轻易上无锁。

3.3 中断安全设计

嵌入式系统里最常见的场景：中断服务函数往队列里写数据，主循环从队列里读数据。这时候锁可能用不了（中断里不能阻塞），怎么办？

我的做法是：关中断保护临界区。

// 中断安全的入队操作
int irq_safe_enqueue(ring_queue_t *q, uint32_t data) {
    int ret;
    uint32_t flags = disable_interrupts();  // 关中断
    ret = ring_enqueue(q, data);
    restore_interrupts(flags);              // 恢复中断
    return ret;
}

关中断的时间一定要短，否则会影响系统实时性。我一般控制在几十个指令周期内，如果队列操作复杂，就考虑用双缓冲或者 DMA。

四、三种队列的对比与选型

特性 环形队列（数组） 链式队列 无锁队列
内存分配 静态，编译时确定 动态，运行时分配 静态
性能 高，O(1) 中等，有 malloc 开销 极高，无锁竞争
线程安全 需加锁或关中断 需加锁 原子操作，无锁
适用场景 消息大小固定，数量已知 消息大小/数量不确定 高性能、中断上下文
风险点 缓冲区溢出 内存碎片、分配失败 ABA 问题、实现复杂

选型其实没那么复杂。你想想看：如果项目里消息类型不超过 5 种，大小固定，用环形队列准没错。如果消息类型多变，比如网络协议栈，链式队列更灵活。至于无锁队列，嗯，除非你真的很懂内存模型，否则慎用。

五、知识体系总览

下面这张图是我自己总结的队列选型决策流程，每次做新项目我都会拿出来看一眼。

消息队列选型决策树 消息队列选型 消息大小/数量固定？ 消息大小/数量可变？ 环形队列（数组） 需要线程安全？ 加互斥锁 关中断保护 链式队列 性能要求极高？ 加锁 + 内存池 无锁队列（慎用） 注：实际项目中需结合 CPU 架构、RTOS 特性、中断优先级综合考量

说实话，队列的实现本身并不难，难的是在真实项目中做出正确的权衡。我见过有人为了炫技用无锁队列，结果 ABA 问题导致数据错乱，排查了整整一周。也见过有人不管三七二十一全用环形队列，结果消息类型一多，代码改得面目全非。

我的建议：从最简单的方案开始。先用环形队列 + 关中断，跑通功能。如果性能不够，再考虑加锁或无锁。不要一开始就上复杂方案，那是给自己挖坑。

好了，队列的数据结构就聊到这里。记住一句话：没有最好的队列，只有最合适的队列。 选型的时候多想想你的实际场景——中断频率多高？消息大小多少？内存够不够？把这些想清楚了，代码自然就写对了。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →