第15章并发数据结构

课程目录第15章并发数据结构

并发数据结构：从锁争抢到无锁艺术

说实话，并发数据结构这块，是我当年从「会用锁」到「懂并发」的一道分水岭。你想想看，平时写个单线程的队列、栈，闭着眼睛都能搞定。但一旦扔到多线程环境里，各种诡异问题就冒出来了——数据错乱、死锁、性能骤降……我早期有个项目，就因为一个线程安全队列没写好，线上服务每隔几小时就卡死一次，排查了整整三天。

嗯，今天我们就来聊聊并发数据结构。我会把线程安全队列、栈、哈希表，以及无锁队列的实现思路，掰开揉碎了讲清楚。

为什么普通数据结构在多线程下会「翻车」？

先看个最简单的例子。一个单链表的头插操作：

// 普通单链表头插
void push_front(node_t* new_node) {
    new_node->next = head;
    head = new_node;
}

两个线程同时执行这段代码，会发生什么？

线程A读到了head，线程B也读到了head。A把new_node->next指向head，然后更新head。B也做同样的事。结果呢？B的更新把A的覆盖了。A插入的那个节点，永远丢失了。

这就是典型的「读-改-写」竞争。我当年第一次遇到这种bug时，还以为是编译器优化搞的鬼……后来才明白，这是并发编程最基础的坑。

线程安全队列：生产者-消费者的核心

队列是最常用的并发数据结构。生产者往里放，消费者往外取。实现方式有很多种，我挑两种最典型的讲。

基于互斥锁的简单队列

#include <pthread.h>
#include <stdbool.h>

typedef struct node {
    void* data;
    struct node* next;
} node_t;

typedef struct {
    node_t* head;
    node_t* tail;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} queue_t;

void queue_init(queue_t* q) {
    q->head = q->tail = NULL;
    pthread_mutex_init(&q->mutex, NULL);
    pthread_cond_init(&q->cond, NULL);
}

void queue_enqueue(queue_t* q, void* data) {
    node_t* node = malloc(sizeof(node_t));
    node->data = data;
    node->next = NULL;

    pthread_mutex_lock(&q->mutex);
    if (q->tail == NULL) {
        q->head = q->tail = node;
    } else {
        q->tail->next = node;
        q->tail = node;
    }
    pthread_cond_signal(&q->cond);
    pthread_mutex_unlock(&q->mutex);
}

void* queue_dequeue(queue_t* q) {
    pthread_mutex_lock(&q->mutex);
    while (q->head == NULL) {
        pthread_cond_wait(&q->cond, &q->mutex);
    }
    node_t* node = q->head;
    void* data = node->data;
    q->head = node->next;
    if (q->head == NULL) {
        q->tail = NULL;
    }
    pthread_mutex_unlock(&q->mutex);
    free(node);
    return data;
}

这个实现很直观。入队加锁，出队加锁，用条件变量处理队列为空的情况。我在一个日志收集系统里用过这种模式，吞吐量大概每秒几万条，够用。

注意：条件变量一定要配合while循环使用，不能用if。为什么？因为pthread_cond_wait可能被「虚假唤醒」。我早期就吃过这个亏，队列为空时消费者被唤醒，结果读到了NULL指针，程序直接崩了。

性能优化：批量操作与双缓冲

如果队列的并发压力很大，锁的争抢会成为瓶颈。我有个项目需要每秒处理百万级消息，上面的实现直接扛不住。怎么办？

一个思路是「批量出队」。消费者一次取出多个元素，减少锁的获取次数：

// 批量出队：一次取出所有元素
void* queue_dequeue_batch(queue_t* q, void** batch, int* count) {
    pthread_mutex_lock(&q->mutex);
    if (q->head == NULL) {
        pthread_mutex_unlock(&q->mutex);
        *count = 0;
        return NULL;
    }
    // 把整个链表摘下来
    node_t* local_head = q->head;
    node_t* local_tail = q->tail;
    q->head = q->tail = NULL;
    pthread_mutex_unlock(&q->mutex);

    // 在锁外遍历链表，填充batch数组
    int i = 0;
    node_t* cur = local_head;
    while (cur && i < MAX_BATCH) {
        batch[i++] = cur->data;
        cur = cur->next;
    }
    *count = i;
    // 释放节点内存（也可以在锁外做）
    // ...
}

这个技巧叫「锁外处理」。把数据从共享结构中摘下来，然后释放锁，再慢慢处理。我做过测试，批量出队能把吞吐量提升3-5倍。

线程安全栈：后进先出的并发挑战

栈比队列简单一些，因为只有一个出入口。但并发场景下，同样有坑。

typedef struct {
    node_t* top;
    pthread_mutex_t mutex;
} stack_t;

void stack_push(stack_t* s, void* data) {
    node_t* node = malloc(sizeof(node_t));
    node->data = data;
    pthread_mutex_lock(&s->mutex);
    node->next = s->top;
    s->top = node;
    pthread_mutex_unlock(&s->mutex);
}

void* stack_pop(stack_t* s) {
    pthread_mutex_lock(&s->mutex);
    if (s->top == NULL) {
        pthread_mutex_unlock(&s->mutex);
        return NULL;
    }
    node_t* node = s->top;
    void* data = node->data;
    s->top = node->next;
    pthread_mutex_unlock(&s->mutex);
    free(node);
    return data;
}

这个实现看起来没问题，对吧？但注意一个细节：free(node) 是在锁外执行的。为什么？因为内存释放可能比较耗时，如果在锁内做，会延长其他线程的等待时间。

小技巧：锁的粒度要尽量小。只保护共享数据的读写，不保护内存分配和释放。这个原则叫「锁外化」，能显著提升并发性能。

线程安全哈希表：读多写少的优化

哈希表比队列和栈复杂得多。因为哈希表涉及桶数组、哈希函数、冲突解决、动态扩容……每个环节都要考虑并发安全。

我参与过一个配置中心项目，哈希表是核心数据结构。配置数据的特点是：读操作远多于写操作。针对这种场景，可以用「读写锁」来优化：

#include <pthread.h>

#define BUCKET_COUNT 1024

typedef struct entry {
    char* key;
    char* value;
    struct entry* next;
} entry_t;

typedef struct {
    entry_t* buckets[BUCKET_COUNT];
    pthread_rwlock_t rwlock;
} hashtable_t;

void hashtable_init(hashtable_t* ht) {
    memset(ht->buckets, 0, sizeof(ht->buckets));
    pthread_rwlock_init(&ht->rwlock, NULL);
}

char* hashtable_get(hashtable_t* ht, const char* key) {
    unsigned int idx = hash(key) % BUCKET_COUNT;
    pthread_rwlock_rdlock(&ht->rwlock);
    entry_t* e = ht->buckets[idx];
    while (e) {
        if (strcmp(e->key, key) == 0) {
            char* val = strdup(e->value);
            pthread_rwlock_unlock(&ht->rwlock);
            return val;
        }
        e = e->next;
    }
    pthread_rwlock_unlock(&ht->rwlock);
    return NULL;
}

void hashtable_put(hashtable_t* ht, const char* key, const char* value) {
    unsigned int idx = hash(key) % BUCKET_COUNT;
    pthread_rwlock_wrlock(&ht->rwlock);
    // 查找是否已存在，存在则更新，不存在则插入
    // ...
    pthread_rwlock_unlock(&ht->rwlock);
}

读写锁的好处是：多个读线程可以同时访问，互不阻塞。只有写线程才会阻塞所有其他线程。在配置中心这种场景下，读操作占99%以上，性能提升非常明显。

但要注意：读写锁不是银弹。如果写操作频繁，或者写操作耗时很长，读线程也会被频繁阻塞。我见过有人把读写锁用在高频交易系统里，结果性能还不如普通互斥锁……

无锁队列：CAS与ABA问题

好了，终于到了最硬核的部分——无锁编程。无锁队列的核心思想是：不用锁，而是用原子操作（CAS）来保证并发安全。

先看一个最简单的无锁栈（Treiber栈）：

#include <stdatomic.h>

typedef struct node {
    void* data;
    struct node* next;
} node_t;

typedef struct {
    atomic_uintptr_t top;
} lockfree_stack_t;

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

node_t* lockfree_pop(lockfree_stack_t* s) {
    node_t* old_top;
    node_t* new_top;
    do {
        old_top = (node_t*)atomic_load(&s->top);
        if (old_top == NULL) return NULL;
        new_top = old_top->next;
    } while (!atomic_compare_exchange_weak(
        &s->top, (uintptr_t*)&old_top, (uintptr_t)new_top));
    return old_top;
}

这个实现看起来很简洁，对吧？但有个大坑——ABA问题。

什么是ABA问题？线程A读取top为节点X，然后被调度走了。线程B把X弹出，又压入一个节点Y，再压入X（注意，X的内存地址没变，但内容变了）。线程A恢复执行，CAS发现top还是X，就认为没变化，于是把top更新为X->next。但此时X->next已经被线程B改过了，数据就乱了。

我当年第一次实现无锁队列时，就被ABA问题坑惨了。排查了整整两天，最后用「带标记的指针」才解决：

// 用64位指针+标记位
// 低48位存地址，高16位存版本号
typedef struct {
    uintptr_t ptr;  // 实际是 (node_t* | version)
} tagged_ptr_t;

// CAS时同时比较地址和版本号
// 每次pop操作都递增版本号
// 这样即使地址相同，版本号不同，CAS也会失败

嗯，无锁编程就是这样——看起来简单，但细节里全是魔鬼。

知识体系总览

下面这张图，是我梳理的并发数据结构核心脉络：

如何选择？我的经验总结

讲了这么多，到底该用哪种？我根据实际项目经验，给个参考：

场景	推荐方案	理由
生产者-消费者，吞吐量<10万/秒	互斥锁队列	实现简单，够用
生产者-消费者，吞吐量>100万/秒	无锁队列	避免锁争抢，性能高
读多写少（如配置中心）	读写锁哈希表	读操作不互斥
写操作频繁	分段锁或无锁	读写锁退化为互斥锁
栈结构，并发压力小	互斥锁栈	实现简单，够用
栈结构，并发压力大	无锁栈（注意ABA）	性能好，但实现复杂

核心原则：能用锁解决的问题，先用锁解决。无锁编程是最后的手段——它带来的性能提升，往往伴随着复杂度的指数级增长。我见过太多人为了炫技而用无锁，结果代码bug一堆，最后又改回有锁版本。

嗯，并发数据结构这块，今天就聊到这儿。记住一句话：没有银弹。理解每种方案的优缺点，根据实际场景做选择，这才是工程师该有的思维方式。

📚 课程目录

并发数据结构：从锁争抢到无锁艺术

为什么普通数据结构在多线程下会「翻车」？

线程安全队列：生产者-消费者的核心

基于互斥锁的简单队列

性能优化：批量操作与双缓冲

线程安全栈：后进先出的并发挑战

线程安全哈希表：读多写少的优化

无锁队列：CAS与ABA问题

知识体系总览

如何选择？我的经验总结