第21章多线程测试与验证

课程目录第21章多线程测试与验证

第21章多线程测试与验证：确定性测试、压力测试、线程安全测试框架、覆盖率分析

说实话，写多线程代码最怕什么？不是死锁，不是数据竞争，而是——这次跑对了，下次跑就崩了。我早年在一个嵌入式项目里，有个bug只在每月第三周的周三下午出现，查了两个月才发现是线程调度时序的巧合。从那以后，我养成了一个习惯：测试多线程，得像审犯人一样，用尽各种手段逼它现形。

这一章，我们就来聊聊怎么系统地“拷问”你的多线程代码。我会把压箱底的经验掏出来，包括确定性测试、压力测试、线程安全测试框架，还有覆盖率分析。

  核心观点： 多线程测试不是“跑一跑没问题就行”，而是要用确定性手段复现问题，用压力测试暴露隐患，用框架固化验证流程，用覆盖率量化测试质量。

21.1 确定性测试：让非确定性变得可控

多线程最大的敌人就是非确定性。同样的代码，这次线程A先跑，下次线程B先跑，结果可能天差地别。确定性测试，说白了就是人为控制线程的执行顺序，让每次测试都走同一条路径。

21.1.1 手动注入调度点

我常用的手法是在关键操作前后加sleep()或yield()，强制让线程在特定点切换。比如测试一个共享队列：

// 测试：生产者先放，消费者后取
void test_producer_first() {
    Queue q;
    init_queue(&q);
    
    // 生产者线程
    pthread_t prod;
    pthread_create(&prod, NULL, producer, &q);
    usleep(100);  // 确保生产者先跑
    
    // 消费者线程
    pthread_t cons;
    pthread_create(&cons, NULL, consumer, &q);
    
    // ... 验证结果
}

嗯，这方法很土，但很有效。我在调试一个双缓冲区的bug时，就是靠这种“人工红绿灯”把问题复现出来的。

21.1.2 使用线程检查器（ThreadSanitizer）

手动调度毕竟费劲。现代工具链提供了更好的选择——ThreadSanitizer（TSan）。它在编译时插桩，运行时检测数据竞争。

// 编译时加上 -fsanitize=thread
// gcc -fsanitize=thread -g -o test test.c -lpthread

#include <pthread.h>
#include <stdio.h>

int counter = 0;  // 共享变量，没加锁

void* worker(void* arg) {
    for (int i = 0; i < 100000; i++) {
        counter++;  // 数据竞争！
    }
    return NULL;
}

int main() {
    pthread_t t1, t2;
    pthread_create(&t1, NULL, worker, NULL);
    pthread_create(&t2, NULL, worker, NULL);
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    printf("counter = %d\n", counter);
    return 0;
}

运行后，TSan会直接告诉你：WARNING: ThreadSanitizer: data race，并给出调用栈。这比你自己猜快多了。

我的习惯： 每次提交多线程代码前，先用TSan跑一遍单元测试。虽然会慢一些，但能拦住90%的数据竞争。

21.2 压力测试：把系统逼到极限

确定性测试能发现逻辑错误，但并发压力下的资源竞争、死锁、饥饿，往往要在高负载下才会暴露。压力测试就是模拟极端场景，看看你的代码会不会“扛不住”。

21.2.1 经典压力测试模式

我一般会写一个压力测试框架，核心思路是：启动大量线程，反复执行同一操作，持续一段时间。

// 压力测试模板
void stress_test(void* (*func)(void*), void* arg, 
                 int num_threads, int duration_sec) {
    pthread_t* threads = malloc(num_threads * sizeof(pthread_t));
    
    // 启动所有线程
    for (int i = 0; i < num_threads; i++) {
        pthread_create(&threads[i], NULL, func, arg);
    }
    
    // 跑够时间
    sleep(duration_sec);
    
    // 通知停止（用全局标志）
    g_stop_flag = 1;
    
    // 等待结束
    for (int i = 0; i < num_threads; i++) {
        pthread_join(threads[i], NULL);
    }
    
    free(threads);
}

举个例子，测试一个线程安全的哈希表：

void* hash_stress(void* arg) {
    HashTable* ht = (HashTable*)arg;
    while (!g_stop_flag) {
        int key = rand() % 10000;
        int val = rand();
        ht_insert(ht, key, val);   // 插入
        ht_get(ht, key);           // 读取
        ht_delete(ht, key);        // 删除
    }
    return NULL;
}

// 启动32个线程，跑60秒
stress_test(hash_stress, &my_ht, 32, 60);

注意： 压力测试跑完后，一定要检查资源泄漏。我曾经有个bug，线程退出时没释放锁，导致压力测试跑半小时后系统锁死。用valgrind --leak-check=full可以查。

21.2.2 压力测试的“黄金三要素”

要素	说明	我的经验值
线程数	至少是CPU核心数的2-4倍	我一般用4倍核心数，比如8核机器开32线程
持续时间	至少跑30分钟，关键系统跑24小时	线上服务我跑72小时，因为有些bug要很久才触发
操作密度	每个线程每秒执行尽可能多的操作	去掉sleep，用自旋或忙等模拟高负载

21.3 线程安全测试框架：把验证变成自动化

手动测试终究不是长久之计。我们需要一个可重复、可自动化的测试框架。我个人比较喜欢用Check或CUnit，但针对多线程，我会做一些定制。

21.3.1 一个轻量级线程安全测试框架

下面是我自己写的一个小框架，专门用来测试线程安全：

// ts_test.h
#ifndef TS_TEST_H
#define TS_TEST_H

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

typedef struct {
    const char* name;
    void (*setup)(void*);
    void (*teardown)(void*);
    void* (*thread_func)(void*);
    int num_threads;
    int iterations;
} TS_TestCase;

void ts_run_test(TS_TestCase* tc) {
    printf("Running: %s\n", tc->name);
    
    void* ctx = NULL;
    if (tc->setup) {
        ctx = malloc(1024);
        tc->setup(ctx);
    }
    
    pthread_t* threads = malloc(tc->num_threads * sizeof(pthread_t));
    for (int i = 0; i < tc->num_threads; i++) {
        pthread_create(&threads[i], NULL, tc->thread_func, ctx);
    }
    
    for (int i = 0; i < tc->num_threads; i++) {
        pthread_join(threads[i], NULL);
    }
    
    if (tc->teardown) {
        tc->teardown(ctx);
        free(ctx);
    }
    
    free(threads);
    printf("PASS: %s\n", tc->name);
}

#endif

使用示例：

// 测试一个线程安全的计数器
void* counter_test(void* arg) {
    int* counter = (int*)arg;
    for (int i = 0; i < 10000; i++) {
        __sync_fetch_and_add(counter, 1);  // 原子操作
    }
    return NULL;
}

int main() {
    TS_TestCase tc = {
        .name = "atomic_counter_test",
        .setup = NULL,
        .teardown = NULL,
        .thread_func = counter_test,
        .num_threads = 10,
        .iterations = 10000
    };
    
    ts_run_test(&tc);
    return 0;
}

避坑指南： 我曾经在setup里分配了全局资源，但忘记在teardown里释放。结果跑完一个测试，下一个测试就崩了。所以每个测试用例必须独立，setup/teardown要成对出现。

21.4 覆盖率分析：量化你的测试质量

测试跑完了，但你怎么知道哪些代码被测试到了，哪些没有？覆盖率分析就是干这个的。对于多线程代码，我特别关注分支覆盖率和线程交互覆盖率。

21.4.1 使用gcov分析代码覆盖率

GCC自带的gcov工具可以帮你统计每行代码的执行次数。编译时加上-fprofile-arcs -ftest-coverage：

// gcc -fprofile-arcs -ftest-coverage -g -o test test.c -lpthread
// 运行后生成 .gcda 文件
// 用 gcov test.c 查看结果

// test.c
#include <pthread.h>

pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
int shared = 0;

void* writer(void* arg) {
    pthread_mutex_lock(&lock);
    shared = 42;  // 行5
    pthread_mutex_unlock(&lock);
    return NULL;
}

void* reader(void* arg) {
    pthread_mutex_lock(&lock);
    int val = shared;  // 行11
    pthread_mutex_unlock(&lock);
    return NULL;
}

int main() {
    pthread_t t1, t2;
    pthread_create(&t1, NULL, writer, NULL);
    pthread_create(&t2, NULL, reader, NULL);
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    return 0;
}

运行gcov test.c后，会生成test.c.gcov文件，里面每行前面有个数字，表示执行次数。如果某行是#####，说明从来没被执行过。

21.4.2 多线程特有的覆盖率指标

除了代码行覆盖率，我还会关注：

锁覆盖率：每个锁是否在至少一个测试中被加锁/解锁
条件变量覆盖率：wait/signal/broadcast是否都被测试到
竞态路径覆盖率：不同的线程交错顺序是否被覆盖

说实话，最后一个很难用工具自动测量。我自己的做法是手动枚举关键竞态场景，比如：

线程A先加锁，线程B等待
线程B先加锁，线程A等待
两个线程同时尝试加锁
一个线程持有锁时被取消

  我的经验： 覆盖率不是越高越好，但低于80%的代码覆盖率，我是不敢上线的。对于多线程代码，我要求关键路径（加锁、解锁、条件变量）的覆盖率达到100%。

知识体系总览

下面这张图，是我对多线程测试与验证的总结。你可以把它当作一个检查清单：

嗯，这套流程我用了好几年，帮我在多个项目里提前发现了隐患。记住：多线程的bug不会消失，只会潜伏。你测试得越狠，线上睡得越安稳。

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

📚 课程目录