手撕环形队列系列二：无锁实现高并发

实力程序员

关注

发布于: 2021 年 07 月 30 日

本文是手撕环形队列系列的第二篇，之前的文章链接如下：

《手撕环形队列》

前面文章介绍的是一个比较基本的环形队列，能够在多线程中使用，但有一个前提：

任意时刻，生产者和消费者最多都只能有一个。

也就是说，如果有多个生产者要并发向队列中写入，需要在外部进行加锁或其它方式的并发控制，保证任意时刻最多只有一个生产者真正向环形队列进行写入。同样的，多个消费者要从队列中读取进行消费，也需要在外部进行加锁或其它方式的并发控制，保证任意时刻最多只有一个消费者从环形队列进行读取。

本文的内容，就是介绍如何能够支持多线程场景下，多生产者并发写入、多消费者并发读取，完全由环形队列内部来解决，无需外部做任何额外的控制。并且，使用无锁的技术来实现，从而避免加锁解锁这种重操作对性能的影响。

无锁数据结构中，主要的技术实现手段是使用 cpu 的原子指令。介绍原子指令之前，先介绍一下没有原子指令的情况下会有什么问题。

通常我们在程序源码中写的语句，编译为二进制后，代码中的一行文本语句会变成二进制的多条汇编指令，因此这一行文本语句 cpu 执行时就不是原子的。多行文本语句，就更不是原子的了。多线程并发执行这些文本语句时，对应的多行汇编语句会在多个 cpu 核上同时执行，无法保证他们之间的执行先后顺序关系。在多线程同时读写一个共享数据时，会发生各种误判，导致错误的结果。

以环形队列为例，来说明这个问题：

环形队列为初始状态，队列为空。两个生产者线程都要向队列进行写入，都调用 ring_queue_push()方法。这个方法的函数实现中，producer1 线程读取 tail 为 0，producer2 线程也读取到 tail 为 0。然后 producer1 向位置 0 写入数据，然后把 tail 增 1，tail 变为 1。

producer2 也向位置 0 写入数据，然后把 tail 增 1. tai 增加 1 的过程：

tail = tail + 1;

由于 producer2 初始读取的 tail 值为 0，这个 cpu core 可能意识不到 tail 已经被别的线程修改了，因此还认为 tail 是 0，因此最终

tail = 0 + 1 = 1;

最终的结果，producer2 把 producer1 的数据给覆盖了(数据丢了)，但两个 ring_queue_push()函数调用都返回成功了。这是一个严重的 Bug！

实际多线程环境中，各个 cpu 之间的代码执行时序都是不同的，因此没有任何防护的情况下，对同样的内存位置写入、对同一个变量的并发读和并发写，都会产生严重的 Bug。

为了解决这些问题，原子指令闪亮登场了！

用这些指令，对数据的操作在多 cpu 的情况下也是原子性的。所谓原子性，就是作为执行的最小单位，不能再分割。cpu core 要么执行了这个指令，要么还没执行这个指令。不会出现在一个 cpu core 执行这个指令一半的时候，另外一个 cpu core 开始执行这个指令的情况。

通过正确使用 cpu 的原子指令，能够有效解决多线程并发中的各种问题。

在解决多线程并发问题，常规的方法是用 mutex、semaphore、condvar 等，这些可以理解为粗粒度锁，使用简单，适用范围广，但性能较差。

cpu 的原子指令，是 cpu 指令级的细粒度锁，性能非常高，但设计起来复杂。

各种操作系统、开发语言中都提供了对 cpu 原子指令的包装函数，因此不需要我们手写汇编指令。

以 gcc 为例，gcc 提供了一系列 builtin 的原子函数，比如今天我们要用的：

bool __sync_bool_compare_and_swap(type *ptr, type oldval， type newval);

这个函数，会将 ptr 指向内存中的值，与 oldval 比较，如果相等，则把 ptr 执行内存的值修改为 newval. 整个比较和修改的全过程，要以原子方式完成。如果比较相等，并且修改成功，则返回 true。其它情况都返回 false。

这个函数，也叫 cas，取的是 compare and swap 三个单词的首字母缩写。

我们用原子指令，来增强一下环形队列，实现多生产多消费者并发读写。思路如下：

对于写入，每个 producer 必须先获得写锁。成功获得写锁之后，写入数据，将 tail 移动到下一个位置，最后释放写锁。

对于读取，每个 consumer 必须先获得读锁。成功获得读锁之后，读取数据，将 head 移动到下一个位置，最后释放读锁。

整个思路，与传统通过 mutex 控制对共享数据的读写是完全一样的，只是技术实现上我们用原子指令来实现，这种实现方式叫无锁数据结构。

另外，需要说明的是：

对于 head 和 tail 这样的变量，由于多个线程会并发读写，因此我们需要用 volatile 来修饰它们，不让 cpu core 缓存它们，避免读到旧数据。

无锁环形队列，支持多生产者多消费者并发读写，用 C 语言实现的源码如下：

// ring_queue.h#ifndef RING_QUEUE_H#define RING_QUEUE_H
typedef struct ring_queue_t {    char* pbuf;    int item_size;    int capacity;
    volatile int write_flag;    volatile int read_flag;
    volatile int head;    volatile int tail;    volatile int same_cycle;} ring_queue_t;
int ring_queue_init(ring_queue_t* pqueue, int item_size, int capacity);void ring_queue_destroy(ring_queue_t* pqueue);int ring_queue_push(ring_queue_t* pqueue, void* pitem);int ring_queue_pop(ring_queue_t* pqueue, void* pitem);int ring_queue_is_empty(ring_queue_t* pqueue);int ring_queue_is_full(ring_queue_t* pqueue);
#endif

复制代码

// ring_queue.c#include "ring_queue.h"#include <stdlib.h>#include <string.h>
#define CAS(ptr, old, new) __sync_bool_compare_and_swap(ptr, old, new)
int ring_queue_init(ring_queue_t* pqueue, int item_size, int capacity) {    memset(pqueue, 0, sizeof(*pqueue));    pqueue->pbuf = (char*)malloc(item_size * capacity);    if (!pqueue->pbuf) {        return -1;    }
    pqueue->item_size = item_size;    pqueue->capacity = capacity;    pqueue->same_cycle = 1;    return 0;}
void ring_queue_destroy(ring_queue_t* pqueue) {    free(pqueue->pbuf);    memset(pqueue, 0, sizeof(*pqueue));}
int ring_queue_push(ring_queue_t* pqueue, void* pitem) {    // try to set write flag    while (1) {        if (ring_queue_is_full(pqueue)) {            return -1;        }
        if (CAS(&pqueue->write_flag, 0, 1)) {   // set write flag successfully            break;        }    }
    // push data    memcpy(pqueue->pbuf + pqueue->tail * pqueue->item_size, pitem, pqueue->item_size);    pqueue->tail = (pqueue->tail + 1) % pqueue->capacity;    if (0 == pqueue->tail) {    // a new cycle        pqueue->same_cycle = 0;     // tail is not the same cycle with head    }
    // reset write flag    CAS(&pqueue->write_flag, 1, 0);
    return 0;}
int ring_queue_pop(ring_queue_t* pqueue, void* pitem) {    // try to set read flag    while (1) {        if (ring_queue_is_empty(pqueue)) {            return -1;        }
        if (CAS(&pqueue->read_flag, 0, 1)) {    // set read flag successfully            break;        }    }
    // read data    memcpy(pitem, pqueue->pbuf + pqueue->head * pqueue->item_size, pqueue->item_size);    pqueue->head = (pqueue->head + 1) % pqueue->capacity;    if (0 == pqueue->head) {        pqueue->same_cycle = 1;     // head is now the same cycle with tail    }
    // reset read flag    CAS(&pqueue->read_flag, 1, 0);
    return 0;}
int ring_queue_is_empty(ring_queue_t* pqueue) {    return (pqueue->head == pqueue->tail) && pqueue->same_cycle;}
int ring_queue_is_full(ring_queue_t* pqueue) {    return (pqueue->head == pqueue->tail) && !pqueue->same_cycle;}

复制代码

我的微信号是实力程序员，欢迎大家转发至朋友圈，分享给更多的朋友。

发布于: 2021 年 07 月 30 日阅读数: 11

原文链接:【http://xie.infoq.cn/article/14462628f5d51363d467c2537】。

实力程序员

关注

实力程序员，用实力说话！ 2021.05.24 加入

超过20年一线产品研发和技术管理的实力程序员

发布

暂无评论

创作场景

手撕环形队列系列二：无锁实现高并发

实力程序员

评论