一、宏与多线程：线程局部存储宏（__thread），原子操作宏，锁封装宏，实战：无锁队列宏

多线程编程，说白了就是让程序同时干好几件事。但问题来了——多个线程同时访问同一份数据，很容易出乱子。我早年做嵌入式项目时，就吃过这个亏。一个全局变量被两个线程同时修改，结果数据错乱，查了三天才找到原因。从那以后，我对多线程的宏封装就格外上心。

今天咱们聊聊C语言里，怎么用宏来搞定线程局部存储、原子操作、锁封装，最后再撸一个无锁队列的宏实现。嗯，内容有点干，但都是实战经验。

1.1 线程局部存储宏（__thread）

先说说线程局部存储。你想想看，有时候我们希望每个线程都有自己的私有数据，比如线程ID、错误码、缓存区。这时候用全局变量肯定不行，因为所有线程共享。用局部变量？函数返回就没了。怎么办？

GCC和Clang提供了__thread关键字。配合宏定义，用起来特别顺手。

#define THREAD_LOCAL __thread

// 每个线程独立的错误码
THREAD_LOCAL int g_thread_errno = 0;

// 线程私有缓冲区
THREAD_LOCAL char g_thread_buf[256];

我在项目中遇到过一个问题：某个日志模块，多个线程同时写日志，结果日志内容互相穿插。后来我用__thread给每个线程分配了一个独立的日志缓冲区，问题就解决了。

小提示：__thread 只能修饰全局变量或静态局部变量，不能修饰函数参数或自动变量。另外，它不能和 extern 一起用，除非变量定义在同一编译单元。

不过要注意，__thread不是标准C，是GCC扩展。MSVC用的是__declspec(thread)。为了跨平台，可以这样封装：

#if defined(__GNUC__) || defined(__clang__)
  #define THREAD_LOCAL __thread
#elif defined(_MSC_VER)
  #define THREAD_LOCAL __declspec(thread)
#else
  #error "Unsupported compiler for thread local storage"
#endif

1.2 原子操作宏

原子操作，说白了就是不可分割的操作。比如对一个整数做自增，在多线程环境下，如果不加保护，可能两个线程同时读、同时写，结果少加了一次。

C11标准引入了stdatomic.h，但嵌入式环境很多不支持。所以咱们还是用GCC内置的原子操作宏。

// 原子自增
#define ATOMIC_INC(ptr)     __sync_fetch_and_add((ptr), 1)

// 原子自减
#define ATOMIC_DEC(ptr)     __sync_fetch_and_sub((ptr), 1)

// 原子交换
#define ATOMIC_SWAP(ptr, val)   __sync_lock_test_and_set((ptr), (val))

// 比较并交换（CAS）
#define ATOMIC_CAS(ptr, old, new)   __sync_bool_compare_and_swap((ptr), (old), (new))

我记得有一次做网络协议栈，多个线程同时操作一个引用计数字段。用普通变量，程序跑着跑着就崩了。换成原子操作宏，稳如老狗。

注意：__sync_* 系列函数在GCC 4.7+中可用，但新版本推荐使用 __atomic_* 系列。如果你用GCC 4.8以上，建议换用 __atomic_*。

新版本的原子宏可以这样写：

#define ATOMIC_LOAD(ptr)        __atomic_load_n((ptr), __ATOMIC_SEQ_CST)
#define ATOMIC_STORE(ptr, val)  __atomic_store_n((ptr), (val), __ATOMIC_SEQ_CST)
#define ATOMIC_FETCH_ADD(ptr, val)  __atomic_fetch_add((ptr), (val), __ATOMIC_SEQ_CST)

这里__ATOMIC_SEQ_CST是内存序，表示顺序一致性。如果你对性能有极致要求，可以换成__ATOMIC_RELAXED，但后果自负——我一般不这么干。

1.3 锁封装宏

锁是保护共享资源的经典手段。但每次写锁代码，都要手动初始化、加锁、解锁、销毁，烦不烦？用宏封装一下，清爽多了。

#include <pthread.h>

// 定义锁变量
#define DEFINE_MUTEX(name)  pthread_mutex_t name = PTHREAD_MUTEX_INITIALIZER

// 加锁
#define LOCK(mutex)         pthread_mutex_lock(&(mutex))

// 解锁
#define UNLOCK(mutex)       pthread_mutex_unlock(&(mutex))

// 自动加锁解锁（RAII风格）
#define SCOPED_LOCK(mutex)  \
    for(int _lock_once = 1; _lock_once; _lock_once = 0) \
        for(pthread_mutex_t *_lock_ptr = &(mutex); _lock_once; _lock_once = 0) \
            for(pthread_mutex_lock(_lock_ptr); _lock_once; pthread_mutex_unlock(_lock_ptr), _lock_once = 0)

这个SCOPED_LOCK宏有点意思。它用三个嵌套的for循环，实现了类似C++ RAII的效果。进入作用域自动加锁，离开作用域自动解锁。我在项目中经常这么用，省心。

用法示例：

DEFINE_MUTEX(g_mutex);

void thread_safe_func(void) {
    SCOPED_LOCK(g_mutex);
    // 这里操作共享资源，安全
    // 函数返回时自动解锁
}
核心要点：锁封装宏的核心思想是「减少重复代码，避免忘记解锁」。尤其是SCOPED_LOCK这种模式，能从根本上防止死锁——只要你不在锁内return或goto。

1.4 实战：无锁队列宏

好了，重头戏来了。无锁队列，说白了就是不用锁，靠原子操作实现线程安全的队列。为什么需要无锁？因为锁会导致线程阻塞，在高并发场景下性能堪忧。

我当年做一个实时数据采集系统，要求每微秒处理上千个数据包。用锁队列，延迟抖动太大，数据经常丢包。后来换成无锁队列，问题就解决了。

下面是一个简单的无锁队列宏实现，基于单生产者单消费者模型（SPSC）：

#include <stdint.h>

// 定义无锁队列结构
#define DEFINE_LOCKFREE_QUEUE(type, size) \
    struct { \
        type buffer[size]; \
        volatile uint32_t head; \
        volatile uint32_t tail; \
    }

// 初始化
#define LOCKFREE_QUEUE_INIT(q)  \
    do { \
        (q).head = 0; \
        (q).tail = 0; \
    } while(0)

// 入队（生产者调用）
#define LOCKFREE_QUEUE_PUSH(q, val) \
    do { \
        uint32_t tail = (q).tail; \
        uint32_t next = (tail + 1) % (sizeof((q).buffer) / sizeof((q).buffer[0])); \
        if (next != (q).head) { \
            (q).buffer[tail] = (val); \
            __sync_synchronize(); \
            (q).tail = next; \
        } \
    } while(0)

// 出队（消费者调用）
#define LOCKFREE_QUEUE_POP(q, pval) \
    do { \
        uint32_t head = (q).head; \
        if (head != (q).tail) { \
            *(pval) = (q).buffer[head]; \
            __sync_synchronize(); \
            (q).head = (head + 1) % (sizeof((q).buffer) / sizeof((q).buffer[0])); \
        } \
    } while(0)

// 判断是否为空
#define LOCKFREE_QUEUE_EMPTY(q) ((q).head == (q).tail)

使用示例：

DEFINE_LOCKFREE_QUEUE(int, 1024) my_queue;

void producer(void) {
    for (int i = 0; i < 1000; i++) {
        LOCKFREE_QUEUE_PUSH(my_queue, i);
    }
}

void consumer(void) {
    int val;
    while (!LOCKFREE_QUEUE_EMPTY(my_queue)) {
        if (LOCKFREE_QUEUE_POP(my_queue, &val)) {
            // 处理val
        }
    }
}
注意：这个实现只适用于单生产者单消费者场景。如果是多生产者或多消费者，需要更复杂的机制（比如CAS操作）。另外，队列大小必须是2的幂次方，否则取模运算会拖慢性能。

为什么这个队列是无锁的？因为入队和出队操作只修改各自的指针（tail和head），没有竞争。唯一需要同步的是内存可见性——用__sync_synchronize()保证一个线程的写入对另一个线程可见。

下面我用一张图来展示无锁队列的核心逻辑：

无锁队列（SPSC）核心逻辑 缓冲区（环形数组） 数据0 数据1 数据2 空 空 空 空 空 空 ... 生产者 消费者 写tail位置 读head位置 tail → 3 head → 0 生产者只修改 tail，消费者只修改 head，无竞争 当 tail + 1 == head 时，队列满 当 tail == head 时，队列空 屏障

从图中可以看到，生产者和消费者各管各的指针，互不干扰。这就是无锁队列能跑得快的根本原因。

实战心得：无锁队列虽然性能好，但调试起来很痛苦。我建议你先用锁队列把功能调通，再换成无锁队列做性能优化。另外，一定要加内存屏障，否则在多核CPU上会出现诡异的数据不一致问题。

好了，这一章的内容就到这里。线程局部存储、原子操作、锁封装、无锁队列——这些宏工具在你写多线程代码时，能帮你省不少事。记住一点：宏只是工具，理解背后的原理才是关键。

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

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