16、序列化与多线程：线程安全的序列化、锁机制、无锁队列中的序列化

多线程环境下的序列化，说白了就是「一群人同时往一个本子上写字，怎么保证字迹不乱」。我早年做嵌入式网关时，就吃过这个亏——两个线程同时序列化同一个结构体，结果数据错位，排查了整整两天。嗯，从那以后，我对线程安全的序列化就格外上心。

为什么多线程序列化容易出问题？

你想想看，序列化操作通常不是原子的。比如你要序列化一个结构体：

// 假设这个结构体被多个线程共享
typedef struct {
    uint32_t id;
    char name[32];
    float value;
} SensorData;

// 序列化函数
void serialize_sensor(const SensorData *data, uint8_t *buffer) {
    memcpy(buffer, &data->id, 4);
    memcpy(buffer + 4, data->name, 32);
    memcpy(buffer + 36, &data->value, 4);
}

如果线程A刚写完id，线程B就把name改了，那序列化出来的结果就是「四不像」。这就是典型的数据竞争问题。

核心原则：序列化操作必须保证「读取原始数据」和「写入目标缓冲区」这两个阶段是原子的，或者至少是互斥的。

方案一：互斥锁 + 序列化

最直接的办法，就是给共享数据加一把锁。我个人习惯用pthread_mutex_t，简单可靠。

#include <pthread.h>

typedef struct {
    SensorData data;
    pthread_mutex_t lock;
} SharedSensor;

void serialize_shared(SharedSensor *shared, uint8_t *buffer) {
    pthread_mutex_lock(&shared->lock);
    // 加锁期间，其他线程不能修改 data
    serialize_sensor(&shared->data, buffer);
    pthread_mutex_unlock(&shared->lock);
}

我在项目中遇到过一个问题：锁的粒度太粗。如果序列化操作很频繁，锁竞争会导致性能急剧下降。这时候可以考虑读写锁：

pthread_rwlock_t rwlock;

void serialize_shared_rw(SharedSensor *shared, uint8_t *buffer) {
    pthread_rwlock_rdlock(&rwlock);  // 读锁，允许多个线程同时读
    serialize_sensor(&shared->data, buffer);
    pthread_rwlock_unlock(&rwlock);
}

void update_shared(SharedSensor *shared, const SensorData *new_data) {
    pthread_rwlock_wrlock(&rwlock);  // 写锁，独占
    shared->data = *new_data;
    pthread_rwlock_unlock(&rwlock);
}

小技巧：如果序列化后的数据要发送到网络或写入文件，建议在锁内只做「内存拷贝」，把耗时的I/O操作放到锁外面。这样可以缩短锁的持有时间。

方案二：无锁队列中的序列化

锁虽然好用，但在高实时性场景下（比如音频处理、工业控制），锁可能引入优先级反转或死锁。这时候无锁队列就派上用场了。

无锁队列的核心思想是：生产者把数据序列化后放入队列，消费者从队列取出并反序列化。双方不直接操作同一块内存，而是通过队列传递「序列化后的字节流」。

// 一个简单的无锁环形队列（单生产者-单消费者）
#define QUEUE_SIZE 64
#define MAX_PACKET_SIZE 128

typedef struct {
    uint8_t buffer[MAX_PACKET_SIZE];
    uint32_t length;
} Packet;

typedef struct {
    Packet slots[QUEUE_SIZE];
    volatile uint32_t head;  // 生产者写入位置
    volatile uint32_t tail;  // 消费者读取位置
} LockFreeQueue;

// 生产者：序列化后入队
int enqueue_serialized(LockFreeQueue *q, const SensorData *data) {
    uint32_t next_head = (q->head + 1) % QUEUE_SIZE;
    if (next_head == q->tail) {
        return -1;  // 队列满
    }
    // 直接在队列槽中序列化
    serialize_sensor(data, q->slots[q->head].buffer);
    q->slots[q->head].length = 40;  // 固定长度
    // 内存屏障，确保写入完成后再更新 head
    __sync_synchronize();
    q->head = next_head;
    return 0;
}

// 消费者：出队并反序列化
int dequeue_deserialize(LockFreeQueue *q, SensorData *out) {
    if (q->tail == q->head) {
        return -1;  // 队列空
    }
    deserialize_sensor(q->slots[q->tail].buffer, out);
    __sync_synchronize();
    q->tail = (q->tail + 1) % QUEUE_SIZE;
    return 0;
}

注意：上面的代码只适用于单生产者-单消费者场景。如果是多生产者，需要用到CAS（Compare-And-Swap）操作。我曾经在项目中用__sync_bool_compare_and_swap实现过多生产者入队，调试起来相当痛苦——无锁编程的坑，比锁还多。

方案三：线程本地存储（TLS）序列化缓冲区

还有一种我比较喜欢的方式：每个线程维护自己的序列化缓冲区。这样根本不需要锁，因为数据是线程私有的。

// 每个线程独立的序列化缓冲区
static __thread uint8_t thread_local_buffer[256];

void thread_safe_serialize(const SensorData *data) {
    serialize_sensor(data, thread_local_buffer);
    // 此时 thread_local_buffer 只属于当前线程
    // 可以安全地发送或写入
}

这种方式在「每个线程独立采集数据，然后序列化上报」的场景下特别好用。但要注意：如果多个线程的数据需要合并，最终还是需要同步机制。

实战对比：三种方案怎么选？

方案 适用场景 性能 复杂度
互斥锁 + 序列化 低频访问，数据共享频繁 中等 低
无锁队列 高频实时，生产者-消费者模型 高 高
线程本地存储 线程独立数据，无需共享 最高 低

我个人建议：能用锁就别用无锁。无锁编程的调试成本太高，除非你确实测出来锁是瓶颈。我曾经在一个项目中用无锁队列替换了互斥锁，性能提升了30%，但代码量翻了一倍，维护成本也上去了。值不值？看你的实时性要求。

知识体系总览

下面这张图总结了本章的核心逻辑：多线程序列化的三种路径，以及它们之间的权衡关系。

多线程序列化方案对比 共享数据需要序列化 方案一：互斥锁 pthread_mutex_lock 序列化 → 解锁 适合低频、共享频繁 方案二：无锁队列 CAS + 环形缓冲区 生产者/消费者分离 适合高频、实时性要求高 方案三：线程本地存储 __thread 关键字 每个线程独立缓冲区 适合线程独立数据 核心原则：序列化操作必须保证数据一致性 无锁 ≠ 无风险，调试成本往往高于锁方案

避坑指南：我曾经在无锁队列中犯过一个低级错误——忘了加内存屏障。结果在ARM处理器上，生产者的写入顺序被CPU重排了，消费者读到了半成品数据。嗯，从那以后我写无锁代码，第一件事就是检查__sync_synchronize()有没有放对位置。

好了，多线程序列化就聊到这里。记住一句话：先保证正确，再追求性能。锁不是魔鬼，无锁也不是银弹。根据你的实际场景，选最合适的方案就好。

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

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