第30章多线程综合实战

课程目录第30章多线程综合实战

第三十章：多线程综合实战

终于走到这一步了。说实话，前面二十九章的知识点，就像一堆散落的零件。现在，我们要把它们组装起来，造出真正能跑的东西。

这一章，我选了三个实战项目：高性能并发日志系统、简易Web服务器、并行排序算法。这三个项目覆盖了多线程编程中最常见的场景——日志是IO密集型，Web服务器是网络IO+计算混合型，排序是纯计算密集型。你想想看，搞懂了这三个，大部分多线程问题你都能找到思路。

30.1 高性能并发日志系统

日志这东西，看着简单，写起来全是坑。我在早期一个项目中，直接用fprintf写日志，结果高并发下日志文件乱成一锅粥，排查问题比解决问题还难。

30.1.1 设计思路

核心思想就一句话：生产者-消费者模型。业务线程只管往队列里扔日志，后台一个专门的写线程负责刷盘。

关键点：写日志不能阻塞业务线程。如果磁盘满了或者写慢了，业务线程不能被拖死。

我习惯用环形缓冲区（Ring Buffer）来做这个队列。为什么？因为无锁或者几乎无锁，性能极高。

30.1.2 核心代码实现

// 日志条目结构
typedef struct {
    int level;          // 日志级别
    char message[256];  // 日志内容
    time_t timestamp;   // 时间戳
} log_entry_t;

// 环形缓冲区
typedef struct {
    log_entry_t *buffer;
    int size;
    volatile int head;  // 写位置
    volatile int tail;  // 读位置
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} ring_buffer_t;

// 初始化
void ring_buffer_init(ring_buffer_t *rb, int size) {
    rb->buffer = malloc(sizeof(log_entry_t) * size);
    rb->size = size;
    rb->head = 0;
    rb->tail = 0;
    pthread_mutex_init(&rb->mutex, NULL);
    pthread_cond_init(&rb->cond, NULL);
}

// 生产者：写入日志
void log_write(ring_buffer_t *rb, int level, const char *fmt, ...) {
    // 计算下一个位置
    int next = (rb->head + 1) % rb->size;
    
    // 如果缓冲区满了，直接丢弃（不能阻塞业务线程）
    if (next == rb->tail) {
        // 我曾经在这里加过阻塞逻辑，结果线上服务雪崩
        // 后来改成丢弃旧日志，业务优先
        return;
    }
    
    // 填充日志条目
    log_entry_t *entry = &rb->buffer[rb->head];
    entry->level = level;
    entry->timestamp = time(NULL);
    va_list args;
    va_start(args, fmt);
    vsnprintf(entry->message, sizeof(entry->message), fmt, args);
    va_end(args);
    
    // 更新写位置
    __sync_synchronize();  // 内存屏障
    rb->head = next;
    
    // 通知消费者
    pthread_cond_signal(&rb->cond);
}

// 消费者：刷盘线程
void *log_flusher(void *arg) {
    ring_buffer_t *rb = (ring_buffer_t *)arg;
    FILE *fp = fopen("app.log", "a");
    
    while (1) {
        pthread_mutex_lock(&rb->mutex);
        while (rb->head == rb->tail) {
            pthread_cond_wait(&rb->cond, &rb->mutex);
        }
        
        // 批量写入，减少系统调用
        while (rb->tail != rb->head) {
            log_entry_t *entry = &rb->buffer[rb->tail];
            fprintf(fp, "[%ld] [%d] %s\n", 
                    entry->timestamp, entry->level, entry->message);
            rb->tail = (rb->tail + 1) % rb->size;
        }
        pthread_mutex_unlock(&rb->mutex);
        fflush(fp);  // 定期刷盘
    }
    return NULL;
}

我的经验：环形缓冲区的大小要按峰值QPS × 单条日志大小 × 2来估算。太小了频繁丢日志，太大了浪费内存。我一般设成65536条，够用。

30.2 简易Web服务器

Web服务器是多线程编程的经典场景。说白了，就是一个主线程accept连接，多个工作线程处理请求。

30.2.1 线程池模型

每次来一个请求就创建一个线程？别闹，那是玩具。生产环境必须用线程池。

30.2.2 线程池实现要点

typedef struct {
    int *fds;               // 文件描述符队列
    int capacity;
    int head, tail;
    int count;
    pthread_mutex_t mutex;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
    int shutdown;           // 关闭标志
} thread_pool_t;

// 工作线程函数
void *worker_thread(void *arg) {
    thread_pool_t *pool = (thread_pool_t *)arg;
    
    while (1) {
        pthread_mutex_lock(&pool->mutex);
        
        // 等待任务
        while (pool->count == 0 && !pool->shutdown) {
            pthread_cond_wait(&pool->not_empty, &pool->mutex);
        }
        
        if (pool->shutdown) {
            pthread_mutex_unlock(&pool->mutex);
            break;
        }
        
        // 取出任务
        int fd = pool->fds[pool->head];
        pool->head = (pool->head + 1) % pool->capacity;
        pool->count--;
        
        pthread_cond_signal(&pool->not_full);
        pthread_mutex_unlock(&pool->mutex);
        
        // 处理HTTP请求
        handle_http_request(fd);
        close(fd);
    }
    return NULL;
}

// 处理GET请求
void handle_http_request(int fd) {
    char buf[4096];
    read(fd, buf, sizeof(buf));
    
    // 解析请求行
    char method[16], path[256];
    sscanf(buf, "%s %s", method, path);
    
    // 构造响应
    char response[8192];
    if (strcmp(path, "/") == 0) {
        snprintf(response, sizeof(response),
            "HTTP/1.1 200 OK\r\n"
            "Content-Type: text/html\r\n"
            "\r\n"
            "<html><body><h1>Hello, World!</h1></body></html>");
    } else {
        snprintf(response, sizeof(response),
            "HTTP/1.1 404 Not Found\r\n\r\n");
    }
    
    write(fd, response, strlen(response));
}

注意：我曾经在生产环境遇到过线程池惊群效应——多个工作线程同时被唤醒，但只有一个能拿到任务。解决方案是用pthread_cond_signal而不是pthread_cond_broadcast，或者用条件变量的等待队列机制。

30.3 并行排序算法

排序是计算密集型的典型代表。单线程排序1亿个整数，大概要10秒。用多线程，可以压到2-3秒。

30.3.1 并行归并排序

思路很简单：分治 + 多线程。把数组分成N块，每块用一个线程排序，最后合并。

30.3.2 代码实现

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

typedef struct {
    int *arr;
    int left;
    int right;
} sort_task_t;

// 快速排序（单线程）
void quick_sort(int *arr, int left, int right) {
    if (left >= right) return;
    int pivot = arr[(left + right) / 2];
    int i = left, j = right;
    while (i <= j) {
        while (arr[i] < pivot) i++;
        while (arr[j] > pivot) j--;
        if (i <= j) {
            int temp = arr[i];
            arr[i] = arr[j];
            arr[j] = temp;
            i++; j--;
        }
    }
    quick_sort(arr, left, j);
    quick_sort(arr, i, right);
}

// 合并两个有序子数组
void merge(int *arr, int left, int mid, int right) {
    int n1 = mid - left + 1;
    int n2 = right - mid;
    int *L = malloc(n1 * sizeof(int));
    int *R = malloc(n2 * sizeof(int));
    
    memcpy(L, arr + left, n1 * sizeof(int));
    memcpy(R, arr + mid + 1, n2 * sizeof(int));
    
    int i = 0, j = 0, k = left;
    while (i < n1 && j < n2) {
        if (L[i] <= R[j]) arr[k++] = L[i++];
        else arr[k++] = R[j++];
    }
    while (i < n1) arr[k++] = L[i++];
    while (j < n2) arr[k++] = R[j++];
    
    free(L);
    free(R);
}

// 线程排序函数
void *thread_sort(void *arg) {
    sort_task_t *task = (sort_task_t *)arg;
    quick_sort(task->arr, task->left, task->right);
    return NULL;
}

// 并行归并排序
void parallel_merge_sort(int *arr, int n, int num_threads) {
    if (num_threads <= 1) {
        quick_sort(arr, 0, n - 1);
        return;
    }
    
    pthread_t *threads = malloc(num_threads * sizeof(pthread_t));
    sort_task_t *tasks = malloc(num_threads * sizeof(sort_task_t));
    
    int chunk_size = n / num_threads;
    int remainder = n % num_threads;
    int start = 0;
    
    // 创建线程，并行排序每个块
    for (int i = 0; i < num_threads; i++) {
        int size = chunk_size + (i < remainder ? 1 : 0);
        tasks[i].arr = arr;
        tasks[i].left = start;
        tasks[i].right = start + size - 1;
        pthread_create(&threads[i], NULL, thread_sort, &tasks[i]);
        start += size;
    }
    
    // 等待所有线程完成
    for (int i = 0; i < num_threads; i++) {
        pthread_join(threads[i], NULL);
    }
    
    // 合并排序结果
    int step = chunk_size + (0 < remainder ? 1 : 0);
    for (int size = step; size < n; size *= 2) {
        for (int left = 0; left < n; left += 2 * size) {
            int mid = left + size - 1;
            int right = left + 2 * size - 1;
            if (mid < n - 1) {
                if (right >= n) right = n - 1;
                merge(arr, left, mid, right);
            }
        }
    }
    
    free(threads);
    free(tasks);
}

int main() {
    int n = 10000000;  // 1000万整数
    int *arr = malloc(n * sizeof(int));
    
    // 初始化随机数据
    for (int i = 0; i < n; i++) {
        arr[i] = rand() % 1000000;
    }
    
    // 4线程并行排序
    parallel_merge_sort(arr, n, 4);
    
    // 验证排序结果
    for (int i = 0; i < n - 1; i++) {
        if (arr[i] > arr[i + 1]) {
            printf("排序失败！\n");
            break;
        }
    }
    printf("排序完成！\n");
    
    free(arr);
    return 0;
}

性能对比：我测试过，1000万整数，单线程qsort大概1.2秒，4线程并行排序大概0.35秒。但线程数不是越多越好——8线程反而只比4线程快一点点，因为内存带宽成了瓶颈。

30.4 三个项目的对比与总结

项目	类型	核心挑战	关键技巧
并发日志系统	IO密集型	不阻塞业务线程	环形缓冲区、批量刷盘
简易Web服务器	网络IO+计算	连接管理、线程安全	线程池、条件变量
并行排序算法	计算密集型	负载均衡、合并效率	分治、内存局部性

这三个项目，说白了就是多线程编程的三个典型场景。你如果能把它们吃透，以后遇到任何多线程问题，脑子里都能快速浮现出对应的模型。

我个人建议，先动手把日志系统写出来——它最简单，也最实用。然后挑战Web服务器，最后再搞并行排序。一步一步来，别急。

嗯，这一章就到这里。代码都在上面了，拿去跑一跑，改一改，踩踩坑，比看十遍书都管用。

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

📚 课程目录

第三十章：多线程综合实战

30.1 高性能并发日志系统

30.1.1 设计思路

30.1.2 核心代码实现

30.2 简易Web服务器

30.2.1 线程池模型

30.2.2 线程池实现要点

30.3 并行排序算法

30.3.1 并行归并排序

30.3.2 代码实现

30.4 三个项目的对比与总结