12、线程安全的数据结构：无锁队列、线程安全栈、线程安全哈希表

说到并发编程，锁是个绕不开的话题。但说实话，锁用多了，性能就下来了。我早年做高并发中间件的时候，就吃过锁的亏——一个简单的队列操作，因为锁竞争，吞吐量直接腰斩。后来我慢慢意识到，有些场景下，我们可以用无锁数据结构来解决问题。

这一章，我们就来聊聊三种常用的线程安全数据结构：无锁队列、线程安全栈、线程安全哈希表。它们各有各的适用场景，也各有各的坑。

12.1 无锁队列（Lock-Free Queue）

无锁队列，说白了就是不用互斥锁，而是靠原子操作来保证线程安全。它的核心思想是：如果操作失败，就重试。这听起来有点笨，但在高并发场景下，它比锁要高效得多。

12.1.1 为什么需要无锁队列？

你想想看，如果两个线程同时往一个队列里 push 数据，用锁的话，一个线程必须等另一个线程释放锁。这个等待时间虽然短，但积少成多，在高并发下就成了瓶颈。无锁队列就不一样了——每个线程都只管做自己的事，失败了就重试，不会阻塞。

我在项目中遇到过这样一个场景：一个实时日志采集系统，每秒要处理几十万条日志。如果用锁队列，CPU 大部分时间都在忙等待，性能根本扛不住。换成无锁队列后，吞吐量提升了 3 倍多。

12.1.2 无锁队列的实现原理

无锁队列通常基于 CAS（Compare-And-Swap）操作实现。CAS 是 CPU 提供的一条原子指令，它的逻辑很简单：

// 伪代码：CAS 操作
bool CAS(int* addr, int expected, int new_value) {
    if (*addr == expected) {
        *addr = new_value;
        return true;
    }
    return false;
}

一个经典的无锁队列实现是 Michael-Scott 队列。它使用两个指针：head 和 tail。push 操作时，线程先读取 tail，然后尝试用 CAS 把新节点链接到 tail 后面。如果 CAS 失败，说明有其他线程抢先了一步，那就重新读取 tail 再试。

#include <stdatomic.h>
#include <stdlib.h>

typedef struct Node {
    void* data;
    struct Node* next;
} Node;

typedef struct {
    Node* head;
    Node* tail;
} LockFreeQueue;

void lfq_init(LockFreeQueue* q) {
    Node* dummy = (Node*)malloc(sizeof(Node));
    dummy->next = NULL;
    q->head = dummy;
    q->tail = dummy;
}

void lfq_push(LockFreeQueue* q, void* data) {
    Node* node = (Node*)malloc(sizeof(Node));
    node->data = data;
    node->next = NULL;

    Node* tail;
    Node* next;
    while (1) {
        tail = atomic_load(&q->tail);
        next = atomic_load(&tail->next);
        if (tail == atomic_load(&q->tail)) {
            if (next == NULL) {
                // 尝试把新节点链接到 tail 后面
                if (atomic_compare_exchange_weak(&tail->next, &next, node)) {
                    break;  // 成功
                }
            } else {
                // tail 落后了，帮它往前移
                atomic_compare_exchange_weak(&q->tail, &tail, next);
            }
        }
    }
    // 移动 tail 指针
    atomic_compare_exchange_weak(&q->tail, &tail, node);
}

核心要点：无锁队列的关键在于「帮助」机制——当一个线程发现 tail 指针落后时，它会主动帮其他线程推进 tail。这避免了线程间的互相等待。

我的经验：无锁队列虽然性能好，但调试起来很痛苦。我曾经花了两天时间，才发现一个 ABA 问题——就是 CAS 操作中，值从 A 变成 B 又变回 A，导致 CAS 误判。解决方法是给指针加上版本号，或者用 128 位的 CAS。

12.2 线程安全栈（Thread-Safe Stack）

栈的特点是后进先出（LIFO）。线程安全栈的实现比队列简单一些，但同样有锁版本和无锁版本。

12.2.1 锁版本的栈

最简单的做法就是给栈加一把互斥锁。push 和 pop 操作都先加锁，再操作。这种做法简单可靠，但性能一般。

#include <pthread.h>
#include <stdlib.h>

typedef struct {
    int* data;
    int top;
    int capacity;
    pthread_mutex_t mutex;
} ThreadSafeStack;

void tss_init(ThreadSafeStack* s, int capacity) {
    s->data = (int*)malloc(capacity * sizeof(int));
    s->top = -1;
    s->capacity = capacity;
    pthread_mutex_init(&s->mutex, NULL);
}

int tss_push(ThreadSafeStack* s, int value) {
    pthread_mutex_lock(&s->mutex);
    if (s->top >= s->capacity - 1) {
        pthread_mutex_unlock(&s->mutex);
        return -1;  // 栈满
    }
    s->data[++s->top] = value;
    pthread_mutex_unlock(&s->mutex);
    return 0;
}

int tss_pop(ThreadSafeStack* s, int* value) {
    pthread_mutex_lock(&s->mutex);
    if (s->top < 0) {
        pthread_mutex_unlock(&s->mutex);
        return -1;  // 栈空
    }
    *value = s->data[s->top--];
    pthread_mutex_unlock(&s->mutex);
    return 0;
}

12.2.2 无锁版本的栈

无锁栈的实现思路和无锁队列类似，也是用 CAS。不过栈的操作更简单——只需要操作一个 top 指针。

#include <stdatomic.h>
#include <stdlib.h>

typedef struct Node {
    int data;
    struct Node* next;
} Node;

typedef struct {
    atomic_uintptr_t top;
} LockFreeStack;

void lfs_init(LockFreeStack* s) {
    atomic_init(&s->top, (uintptr_t)NULL);
}

void lfs_push(LockFreeStack* s, int value) {
    Node* node = (Node*)malloc(sizeof(Node));
    node->data = value;

    Node* old_top;
    do {
        old_top = (Node*)atomic_load(&s->top);
        node->next = old_top;
    } while (!atomic_compare_exchange_weak(&s->top, (uintptr_t*)&old_top, (uintptr_t)node));
}

int lfs_pop(LockFreeStack* s, int* value) {
    Node* old_top;
    Node* new_top;
    do {
        old_top = (Node*)atomic_load(&s->top);
        if (old_top == NULL) {
            return -1;  // 栈空
        }
        new_top = old_top->next;
    } while (!atomic_compare_exchange_weak(&s->top, (uintptr_t*)&old_top, (uintptr_t)new_top));

    *value = old_top->data;
    free(old_top);
    return 0;
}

注意：无锁栈同样有 ABA 问题。上面的代码为了简洁没有处理，实际项目中建议使用带版本号的指针，或者用 hazard pointer 来管理内存回收。

12.3 线程安全哈希表（Thread-Safe Hash Table）

哈希表是日常开发中最常用的数据结构之一。线程安全哈希表的实现策略有很多种，我挑两种最实用的来讲。

12.3.1 细粒度锁哈希表

给整个哈希表加一把大锁，虽然简单，但并发度太低。更好的做法是：每个桶（bucket）单独加锁。这样，不同线程操作不同桶时，完全不会互相干扰。

#include <pthread.h>
#include <stdlib.h>
#include <string.h>

#define BUCKET_COUNT 1024

typedef struct Entry {
    char* key;
    int value;
    struct Entry* next;
} Entry;

typedef struct {
    Entry* head;
    pthread_mutex_t mutex;
} Bucket;

typedef struct {
    Bucket buckets[BUCKET_COUNT];
} ThreadSafeHashTable;

unsigned int hash(const char* key) {
    unsigned int h = 0;
    while (*key) {
        h = h * 31 + *key++;
    }
    return h % BUCKET_COUNT;
}

void tsh_init(ThreadSafeHashTable* ht) {
    for (int i = 0; i < BUCKET_COUNT; i++) {
        ht->buckets[i].head = NULL;
        pthread_mutex_init(&ht->buckets[i].mutex, NULL);
    }
}

int tsh_put(ThreadSafeHashTable* ht, const char* key, int value) {
    unsigned int idx = hash(key);
    Bucket* bucket = &ht->buckets[idx];

    pthread_mutex_lock(&bucket->mutex);
    // 查找是否已存在
    Entry* cur = bucket->head;
    while (cur) {
        if (strcmp(cur->key, key) == 0) {
            cur->value = value;
            pthread_mutex_unlock(&bucket->mutex);
            return 0;  // 更新
        }
        cur = cur->next;
    }
    // 插入新节点
    Entry* new_entry = (Entry*)malloc(sizeof(Entry));
    new_entry->key = strdup(key);
    new_entry->value = value;
    new_entry->next = bucket->head;
    bucket->head = new_entry;
    pthread_mutex_unlock(&bucket->mutex);
    return 1;  // 新增
}

int tsh_get(ThreadSafeHashTable* ht, const char* key, int* value) {
    unsigned int idx = hash(key);
    Bucket* bucket = &ht->buckets[idx];

    pthread_mutex_lock(&bucket->mutex);
    Entry* cur = bucket->head;
    while (cur) {
        if (strcmp(cur->key, key) == 0) {
            *value = cur->value;
            pthread_mutex_unlock(&bucket->mutex);
            return 0;  // 找到
        }
        cur = cur->next;
    }
    pthread_mutex_unlock(&bucket->mutex);
    return -1;  // 未找到
}

细粒度锁的优势：当桶的数量足够多时，不同线程几乎不会竞争同一把锁。理论上，并发度可以接近桶的数量。

12.3.2 无锁哈希表（RCU 方案）

如果你追求极致性能，可以考虑用 RCU（Read-Copy-Update）来实现无锁哈希表。RCU 的核心思想是：读操作不加锁，写操作先复制再更新，最后回收旧数据。

RCU 的实现比较复杂，我简单说一下思路：

  • 读操作：直接读取，不加任何锁。通过内存屏障保证读到的是最新数据。
  • 写操作：先复制一份要修改的数据，在副本上修改，然后用原子操作把新数据替换进去。
  • 回收：等待所有正在读的线程都完成后，再释放旧数据。

我的建议：RCU 方案虽然性能好，但实现难度大，而且对内存管理要求很高。如果你不是特别追求极致性能，细粒度锁哈希表已经够用了。我在项目中一般先用细粒度锁，等性能瓶颈出现时再考虑无锁方案。

12.4 三种数据结构的对比

数据结构 实现难度 读性能 写性能 适用场景
无锁队列 高 高 高 生产者-消费者模型、日志收集
线程安全栈（锁版本） 低 中 中 任务池、资源池
线程安全哈希表（细粒度锁） 中 高 中 缓存、配置存储
无锁哈希表（RCU） 极高 极高 中 读多写少的场景

12.5 知识体系图

下面这张图展示了本章的核心知识结构，方便你快速回顾：

线程安全数据结构知识体系 无锁队列 线程安全栈 线程安全哈希表 CAS 原子操作 Michael-Scott 队列 ABA 问题与版本号 锁版本：互斥锁保护 无锁版本：CAS 操作 top Hazard Pointer 内存管理 细粒度锁：每桶一锁 RCU 无锁方案 读多写少场景优化 选择原则：根据读写比例、并发度、实现成本权衡

12.6 避坑指南

最后，我总结几个常见的坑，都是我曾经踩过的：

  • 内存泄漏：无锁数据结构的内存回收是个大问题。我建议用 hazard pointer 或 epoch-based reclamation，别自己瞎搞。
  • 死锁：细粒度锁哈希表虽然并发度高，但如果你在持有 A 桶锁时去获取 B 桶锁，就可能死锁。解决办法是：按固定顺序加锁。
  • 伪共享：多个线程频繁访问同一缓存行的不同变量，会导致性能骤降。可以用 __attribute__((aligned(64))) 来对齐。
  • 测试不充分：并发 bug 很难复现。我建议用 ThreadSanitizer 工具来检测数据竞争，别靠肉眼找。

嗯，这一章的内容就到这里。线程安全数据结构是并发编程的基石，选对了能省很多事。希望你能在实际项目中多动手，踩踩坑，才能真正理解它们。

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

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