C++高性能计算实战：多线程 SIMD 内存池与并发数据结构

（2）同步机制

• 互斥锁（std::mutex）：保证临界区资源原子访问，避免数据竞争；
• 智能锁（std::lock_guard/std::unique_lock）：自动释放锁，避免手动解锁遗漏；
• 条件变量（std::condition_variable）：实现线程间通信，如生产者 - 消费者模型。

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> task_queue;
bool is_running = true;

// 生产者线程
void producer() {
    for (int i = 0; i < 5; ++i) {
        std::lock_guard<std::mutex> lock(mtx); // 自动加锁/解锁
        task_queue.push(i);
        std::cout << "Produce task: " << i << std::endl;
        cv.notify_one(); // 唤醒一个等待线程
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
    }
}

// 消费者线程
void consumer() {
    while (is_running) {
        std::unique_lock<std::mutex> lock(mtx); // 等待条件满足（队列非空）
        cv.wait(lock, [](){return !task_queue.empty() || !is_running;});
        if (!task_queue.empty()) {
            int task = task_queue.front();
            task_queue.pop();
            std::cout << "Consume task: " << task << std::endl;
        }
    }
}

int main() {
    std::thread prod(producer);
    std::thread cons(consumer);
    prod.join();
    is_running = false;
    cv.notify_one(); // 唤醒消费者线程退出
    cons.join();
    return 0;
}

2.2 C++14/17线程特性增强

（1）C++14特性

• 泛型 lambda：支持自动推导参数类型，简化线程函数编写；
• std::shared_timed_mutex：读写锁，支持多个读线程同时访问，提升读密集场景性能。

#include <iostream>
#include <thread>
#include <shared_mutex>

std::shared_timed_mutex rw_mtx;
int shared_data = 0;

// 读线程（共享锁）
void reader(int id) {
    std::shared_lock<std::shared_timed_mutex> lock(rw_mtx);
    std::cout << "Reader " << id << " read data: " << shared_data << std::endl;
}

// 写线程（独占锁）
void writer() {
    std::unique_lock<std::shared_timed_mutex> lock(rw_mtx);
    shared_data++;
    std::cout << "Writer update data: " << shared_data << std::endl;
}

int main() {
    std::thread r1(reader, 1), r2(reader, 2), w(writer), r3(reader, 3);
    r1.join();
    r2.join();
    w.join();
    r3.join();
    return 0;
}

（2）C++17特性

• std::execution：并行算法库，支持std::sort等算法的并行执行；
• std::jthread：可取消线程，自动 join，避免资源泄漏；
• std::scoped_lock：多锁同时加锁，避免死锁。

#include <iostream>
#include <vector>
#include <algorithm>
#include <execution> // C++17 并行算法头文件
#include <thread>

int main() {
    // 并行排序（C++17）
    std::vector<int> vec(1000000);
    std::generate(vec.begin(), vec.end(), [](){return rand()%10000;});
    // 并行排序（std::execution::par 表示并行执行）
    std::sort(std::execution::par, vec.begin(), vec.end());
    
    // std::jthread 自动 join
    std::jthread t([](std::stop_token st){
        while(!st.stop_requested()){ 
            std::cout << "Jthread running..." << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    });
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    t.request_stop(); // 取消线程
    return 0;
}

2.3 实战：高性能线程池实现

线程池通过复用线程减少线程创建/销毁开销，是高并发场景核心组件，核心设计要点：任务队列、线程数组、同步机制、拒绝策略。

#include <iostream>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <functional>
#include <atomic>

class ThreadPool {
public:
    // 构造函数：初始化线程池
    explicit ThreadPool(size_t thread_num = std::thread::hardware_concurrency()) : is_running_(true) {
        // 线程数默认设为 CPU 核心数
        for (size_t i = 0; i < thread_num; ++i) {
            threads_.emplace_back([this]() { this->worker(); });
        }
    }

    // 析构函数：关闭线程池
    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lock(mtx_);
            is_running_ = false;
        }
        cv_.notify_all(); // 唤醒所有工作线程
        for (auto& t : threads_) {
            if (t.joinable()) {
                t.join();
            }
        }
    }

    // 提交任务
    template<typename F, typename... Args>
    auto submit(F&& f, Args&&... args) -> std::future<decltype(f(args...))> {
        // 封装任务为可调用对象
        using ReturnType = decltype(f(args...));
        auto task = std::make_shared<std::packaged_task<ReturnType()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...));
        std::future<ReturnType> future = task->get_future();
        {
            std::lock_guard<std::mutex> lock(mtx_);
            if (!is_running_) {
                throw std::runtime_error("ThreadPool is stopped");
            }
            task_queue_.emplace([task]() { (*task)(); });
        }
        cv_.notify_one(); // 唤醒一个工作线程
        return future;
    }

private:
    // 工作线程函数
    void worker() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(mtx_); // 等待任务或线程池关闭
                cv_.wait(lock, [this]() { return !is_running_ || !task_queue_.empty(); });
                if (!is_running_ && task_queue_.empty()) {
                    return; // 线程池关闭且无任务，退出
                }
                // 取出任务
                task = std::move(task_queue_.front());
                task_queue_.pop();
            }
            task(); // 执行任务
        }
    }

private:
    std::vector<std::thread> threads_; // 工作线程数组
    std::queue<std::function<void()>> task_queue_; // 任务队列
    std::mutex mtx_; // 互斥锁
    std::condition_variable cv_; // 条件变量
    std::atomic<bool> is_running_; // 线程池运行状态
};

// 测试
int main() {
    ThreadPool pool(4); // 4 线程池
    for (int i = 0; i < 8; ++i) {
        pool.submit([i]() {
            std::cout << "Task " << i << " run in thread " << std::this_thread::get_id() << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }
    return 0;
}

2.4 多线程避坑指南

1. 避免数据竞争：所有共享变量必须通过互斥锁/原子变量访问，禁止裸指针跨线程传递；
2. 避免死锁：遵循'锁顺序一致'原则，或使用 std::scoped_lock 同时加锁多个 mutex；
3. 减少锁开销：缩小临界区范围，读密集场景用读写锁（shared_timed_mutex）；
4. 避免线程过多：线程数建议不超过 CPU 核心数×2（IO 密集型）或 CPU 核心数 +1（计算密集型）；
5. 禁止线程 detach() 后访问其资源：detach() 后线程独立运行，资源可能提前释放。

3. SIMD 指令优化实战

SIMD（单指令多数据）是 CPU 向量计算技术，通过一条指令并行处理多个数据，大幅提升计算密集型任务性能，主流指令集：SSE（128位）、AVX（256位）、AVX-512（512位）。

3.1 SIMD 核心原理与指令集选型

（1）核心原理

SIMD 通过扩展 CPU 寄存器宽度，将多个数据打包到一个向量寄存器中，单次指令完成对所有数据的运算，如 AVX2 寄存器可同时存储 8 个 int32 数据，一次加法指令完成 8 个数据的加法。

（2）指令集选型

• SSE4.2：兼容绝大多数 CPU（Intel/AMD），128 位向量，支持整数/浮点运算；
• AVX2：Intel Haswell 及以后、AMD Ryzen 及以后支持，256 位向量，性能比 SSE 提升一倍；
• AVX-512：Intel Xeon/Core i9 支持，512 位向量，适合超高性能计算场景。

3.2 编译器 Intrinsic 函数使用

编译器提供 SIMD intrinsic 函数（本质是汇编指令的封装），无需直接编写汇编，即可调用 SIMD 指令，常用编译器：GCC/Clang、MSVC。

（1）头文件与编译选项

• SSE：#include <emmintrin.h>（SSE2）、<smmintrin.h>（SSE4.1）；
• AVX：#include <immintrin.h>（AVX/AVX2/AVX-512）；
• 编译选项：GCC/Clang 用 -msse4.2/-mavx2/-mavx512f，MSVC 用 /arch:SSE2/AVX2。

3.3 实战：SIMD 优化数组运算

以'两个 int 数组相加'为例，对比普通实现与 AVX2 优化实现的性能差异。

#include <iostream>
#include <vector>
#include <chrono>
#include <immintrin.h> // AVX2 头文件

// 普通实现：逐元素相加
void add_normal(const std::vector<int>& a, const std::vector<int>& b, std::vector<int>& c) {
    for (size_t i = 0; i < a.size(); ++i) {
        c[i] = a[i] + b[i];
    }
}

// AVX2 优化实现：一次处理 8 个 int（256 位=8×32 位）
void add_avx2(const std::vector<int>& a, const std::vector<int>& b, std::vector<int>& c) {
    size_t i = 0;
    const size_t batch_size = 8; // AVX2 一次处理 8 个 int
    for (; i < a.size() - batch_size + 1; i += batch_size) {
        // 加载 8 个 int 到 AVX2 寄存器
        __m256i vec_a = _mm256_loadu_si256((__m256i*)&a[i]);
        __m256i vec_b = _mm256_loadu_si256((__m256i*)&b[i]);
        // 向量加法：8 个 int 同时相加
        __m256i vec_c = _mm256_add_epi32(vec_a, vec_b);
        // 存储结果到内存
        _mm256_storeu_si256((__m256i*)&c[i], vec_c);
    }
    // 处理剩余元素（不足 8 个）
    for (; i < a.size(); ++i) {
        c[i] = a[i] + b[i];
    }
}

// 性能测试
int main() {
    const size_t size = 10000000; // 1000 万元素
    std::vector<int> a(size, 1), b(size, 2), c(size, 0);
    
    // 测试普通实现
    auto start = std::chrono::high_resolution_clock::now();
    add_normal(a, b, c);
    auto end = std::chrono::high_resolution_clock::now();
    std::cout << "Normal add time: " << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count() << "ms" << std::endl;
    
    // 测试 AVX2 实现
    start = std::chrono::high_resolution_clock::now();
    add_avx2(a, b, c);
    end = std::chrono::high_resolution_clock::now();
    std::cout << "AVX2 add time: " << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count() << "ms" << std::endl;
    
    return 0;
}

编译命令（GCC）：g++ -O3 -mavx2 simd_test.cpp -o simd_test

性能对比：AVX2 优化后性能提升约 3~4 倍（具体取决于 CPU 型号）。

3.4 SIMD 优化注意事项

1. 数据对齐：_mm256_load_si256 要求内存 16/32 字节对齐，未对齐用 _mm256_loadu_si256（u=unaligned）；
2. 数据长度：尽量让数组长度为 SIMD 批次大小的整数倍，减少剩余元素处理；
3. 避免分支：SIMD 指令对分支敏感，分支会破坏并行性，尽量用条件执行指令替代 if-else；
4. 编译器优化：开启 O3 优化，编译器会自动进行部分 SIMD 优化，可通过汇编代码验证；
5. 指令集兼容：根据目标平台选择指令集，避免使用 AVX-512 在老旧 CPU 上运行。

4. 内存池设计与实现

频繁调用 new/delete 会导致内存碎片、性能低下（系统调用开销大），内存池通过预先分配内存块、复用内存，大幅提升内存分配效率，是高性能系统核心组件。

4.1 内存池核心设计思想

1. 预分配：程序启动时分配一大块连续内存（内存池）；
2. 内存块管理：将内存池划分为固定大小或可变大小的内存块，用空闲链表管理；
3. 复用机制：释放内存时不归还系统，而是加入空闲链表，下次分配时直接复用；
4. 内存对齐：保证分配的内存块按指定字节对齐（如 8/16 字节），提升 CPU 访问效率。

4.2 固定大小内存池实现

固定大小内存池适合分配大小固定的对象（如链表节点、任务对象），设计简单、性能最优。

#include <iostream>
#include <cstdlib>
#include <cstdint>

class FixedSizeMemoryPool {
public:
    // 构造函数：初始化内存池
    FixedSizeMemoryPool(size_t block_size, size_t pool_size) : block_size_(align_up(block_size)), pool_size_(pool_size) {
        // 计算总内存大小：块大小 × 块数量
        total_size_ = block_size_ * pool_size_;
        // 分配内存池（从堆上分配连续内存）
        pool_start_ = (char*)std::malloc(total_size_);
        if (!pool_start_) {
            throw std::bad_alloc();
        }
        // 初始化空闲链表（将所有块串联）
        free_list_ = pool_start_;
        char* curr = pool_start_;
        for (size_t i = 0; i < pool_size_ - 1; ++i) {
            // 每个块的头部存储下一个块的地址（空闲时）
            *(char**)curr = curr + block_size_;
            curr += block_size_;
        }
        *(char**)curr = nullptr; // 最后一个块的下一个地址为 null
    }

    // 析构函数：释放内存池
    ~FixedSizeMemoryPool() {
        std::free(pool_start_);
    }

    // 分配内存块
    void* allocate() {
        if (!free_list_) {
            // 空闲链表为空，可扩容或返回 null（此处简单返回 null）
            return nullptr;
        }
        // 取出空闲链表头部块
        void* ptr = free_list_;
        free_list_ = *(char**)free_list_;
        return ptr;
    }

    // 释放内存块
    void deallocate(void* ptr) {
        if (!ptr) return;
        // 检查 ptr 是否在内存池范围内
        if (ptr < pool_start_ || ptr >= pool_start_ + total_size_) {
            throw std::invalid_argument("Invalid pointer to deallocate");
        }
        // 将块插入空闲链表头部
        *(char**)ptr = free_list_;
        free_list_ = (char*)ptr;
    }

private:
    // 内存对齐：将 size 向上对齐到 8 字节（可调整为 16/32 字节）
    size_t align_up(size_t size) {
        const size_t align = 8;
        return (size + align - 1) & ~(align - 1);
    }

private:
    char* pool_start_; // 内存池起始地址
    char* free_list_; // 空闲链表头部
    size_t block_size_; // 单个内存块大小（对齐后）
    size_t pool_size_; // 内存块数量
    size_t total_size_; // 内存池总大小
};

// 测试
int main() {
    // 创建内存池：每个块 32 字节，共 100 个块
    FixedSizeMemoryPool pool(32, 100);
    // 分配 5 个块
    void* p1 = pool.allocate();
    void* p2 = pool.allocate();
    void* p3 = pool.allocate();
    void* p4 = pool.allocate();
    void* p5 = pool.allocate();
    std::cout << "Allocated pointers: " << p1 << " " << p2 << " " << p3 << " " << p4 << " " << p5 << std::endl;
    // 释放 p2 和 p4
    pool.deallocate(p2);
    pool.deallocate(p4);
    // 再次分配，会复用释放的块
    void* p6 = pool.allocate();
    void* p7 = pool.allocate();
    std::cout << "Reallocated pointers: " << p6 << " " << p7 << std::endl;
    return 0;
}

4.3 可变大小内存池优化

可变大小内存池支持分配不同大小的内存块，核心设计：按大小区间划分多个固定大小内存池（如 8B、16B、32B、64B、128B），分配时选择匹配的内存池。

#include <iostream>
#include <vector>
#include <cstdint>
#include <algorithm>

// 固定大小内存池（复用之前的实现）
class FixedSizeMemoryPool {
    // ... 此处省略，与 4.2 实现一致 ...
};

class VariableSizeMemoryPool {
public:
    VariableSizeMemoryPool() {
        // 初始化不同大小的固定内存池：8B、16B、32B、64B、128B，每个池 100 个块
        pool_sizes_ = {8, 16, 32, 64, 128};
        for (size_t size : pool_sizes_) {
            pools_.emplace_back(std::make_unique<FixedSizeMemoryPool>(size, 100));
        }
    }

    // 分配内存
    void* allocate(size_t size) {
        if (size == 0) return nullptr;
        // 找到匹配的内存池（大于等于 size 的最小块大小）
        auto it = std::lower_bound(pool_sizes_.begin(), pool_sizes_.end(), size);
        if (it == pool_sizes_.end()) {
            // 超过最大块大小，直接调用 malloc
            return std::malloc(size);
        }
        // 从对应内存池分配
        size_t pool_size = *it;
        size_t pool_idx = it - pool_sizes_.begin();
        void* ptr = pools_[pool_idx]->allocate();
        if (ptr) return ptr;
        // 内存池满了，扩容（简单扩容 100 个块）
        pools_[pool_idx] = std::make_unique<FixedSizeMemoryPool>(pool_size, 200);
        return pools_[pool_idx]->allocate();
    }

    // 释放内存
    void deallocate(void* ptr, size_t size) {
        if (!ptr) return;
        if (size == 0) return;
        // 找到匹配的内存池
        auto it = std::lower_bound(pool_sizes_.begin(), pool_sizes_.end(), size);
        if (it == pool_sizes_.end()) {
            // 释放 malloc 分配的内存
            std::free(ptr);
            return;
        }
        size_t pool_size = *it;
        size_t pool_idx = it - pool_sizes_.begin();
        // 尝试从内存池释放，失败则调用 free
        try {
            pools_[pool_idx]->deallocate(ptr);
        } catch (const std::invalid_argument&) {
            std::free(ptr);
        }
    }

private:
    std::vector<size_t> pool_sizes_; // 内存池块大小列表
    std::vector<std::unique_ptr<FixedSizeMemoryPool>> pools_; // 多个固定大小内存池
};

// 测试
int main() {
    VariableSizeMemoryPool pool;
    void* p1 = pool.allocate(10); // 匹配 16B 内存池
    void* p2 = pool.allocate(40); // 匹配 64B 内存池
    void* p3 = pool.allocate(200); // 超过 128B，调用 malloc
    std::cout << "Allocated: " << p1 << " " << p2 << " " << p3 << std::endl;
    pool.deallocate(p1, 10);
    pool.deallocate(p2, 40);
    pool.deallocate(p3, 200);
    return 0;
}

4.4 内存池性能对比

通过分配/释放 100 万个 8 字节对象，对比内存池与 new/delete 的性能：

5. 高性能并发数据结构

多线程环境下，普通数据结构（std::vector、std::queue）非线程安全，直接使用会导致数据竞争，高性能并发数据结构需通过锁优化、无锁编程提升并发效率。

5.1 并发数据结构设计原则

1. 最小锁粒度：尽量缩小临界区，避免整个数据结构加锁；
2. 读写分离：读密集场景用读写锁，提升并发读性能；
3. 无锁设计：用 CAS（compare-and-swap）指令实现无锁操作，避免锁开销；
4. 分区锁：将数据结构划分为多个分区，每个分区独立加锁（如 ConcurrentHashMap）。

5.2 实战：线程安全队列

基于 CAS 实现无锁队列（Michael-Scott 队列），适合高并发场景，无锁操作避免上下文切换和锁竞争。

#include <iostream>
#include <atomic>
#include <thread>
#include <vector>

template<typename T>
class LockFreeQueue {
private:
    // 队列节点
    struct Node {
        T data;
        std::atomic<Node*> next;
        Node(const T& data) : data(data), next(nullptr) {}
    };
    std::atomic<Node*> head_; // 队列头部
    std::atomic<Node*> tail_; // 队列尾部

public:
    LockFreeQueue() {
        // 哨兵节点（空节点），简化入队/出队逻辑
        Node* sentinel = new Node(T());
        head_.store(sentinel);
        tail_.store(sentinel);
    }

    ~LockFreeQueue() {
        // 释放所有节点
        while (Node* node = head_.load()) {
            head_.store(node->next.load());
            delete node;
        }
    }

    // 入队
    void enqueue(const T& data) {
        Node* new_node = new Node(data);
        Node* old_tail = nullptr;
        do {
            old_tail = tail_.load(); // 找到尾部节点的 next（应为 null）
        } while (!old_tail->next.compare_exchange_weak(nullptr, new_node, std::memory_order_release, std::memory_order_relaxed));
        // 更新 tail 指针（允许滞后，不影响正确性）
        tail_.compare_exchange_strong(old_tail, new_node);
    }

    // 出队（返回是否成功，数据存入 data）
    bool dequeue(T& data) {
        Node* old_head = nullptr;
        Node* new_head = nullptr;
        do {
            old_head = head_.load();
            new_head = old_head->next.load();
            if (!new_head) {
                return false; // 队列为空
            }
            // 读取数据（需确保节点未被其他线程删除）
            data = new_head->data;
            // 更新 head 指针到新节点（哨兵节点）
        } while (!head_.compare_exchange_weak(
            old_head, new_head, std::memory_order_release, std::memory_order_relaxed));
        delete old_head; // 释放旧哨兵节点
        return true;
    }

    // 判断队列是否为空
    bool empty() const {
        return head_.load()->next.load() == nullptr;
    }
};

// 测试
int main() {
    LockFreeQueue<int> queue;
    const int task_num = 100000;
    std::atomic<int> count(0);
    // 4 个生产者线程入队
    std::vector<std::thread> producers;
    for (int i = 0; i < 4; ++i) {
        producers.emplace_back([&]() {
            for (int j = 0; j < task_num; ++j) {
                queue.enqueue(j);
            }
        });
    }
    // 4 个消费者线程出队
    std::vector<std::thread> consumers;
    for (int i = 0; i < 4; ++i) {
        consumers.emplace_back([&]() {
            int data;
            while (count < task_num * 4) {
                if (queue.dequeue(data)) {
                    count++;
                }
            }
        });
    }
    for (auto& t : producers) t.join();
    for (auto& t : consumers) t.join();
    std::cout << "Total dequeued: " << count << " (expected: " << task_num * 4 << ")" << std::endl;
    return 0;
}

5.3 实战：无锁哈希表

无锁哈希表基于 CAS 指令实现，采用链地址法解决哈希冲突，核心优化：分段锁、原子操作保证并发安全。

#include <iostream>
#include <vector>
#include <atomic>
#include <functional>
#include <utility>

template<typename K, typename V, size_t Capacity = 1024>
class LockFreeHashTable {
private:
    // 哈希表节点
    struct Node {
        K key;
        V value;
        std::atomic<Node*> next;
        Node(K key, V value) : key(key), value(value), next(nullptr) {}
    };
    std::vector<std::atomic<Node*>> table_; // 哈希表数组
    std::hash<K> hash_; // 哈希函数

public:
    LockFreeHashTable() {
        // 初始化哈希表，每个桶为空
        table_.resize(Capacity);
        for (auto& bucket : table_) {
            bucket.store(nullptr);
        }
    }

    ~LockFreeHashTable() {
        // 释放所有节点
        for (size_t i = 0; i < Capacity; ++i) {
            Node* node = table_[i].load();
            while (node) {
                Node* next = node->next.load();
                delete node;
                node = next;
            }
        }
    }

    // 插入/更新键值对
    bool insert(const K& key, const V& value) {
        size_t idx = hash_(key) % Capacity;
        Node* new_node = new Node(key, value);
        Node* old_head = nullptr;
        do {
            old_head = table_[idx].load();
            new_node->next.store(old_head); // CAS 更新桶的头部节点
        } while (!table_[idx].compare_exchange_weak(
            old_head, new_node, std::memory_order_release, std::memory_order_relaxed));
        return true;
    }

    // 查找键值对
    bool find(const K& key, V& value) const {
        size_t idx = hash_(key) % Capacity;
        Node* node = table_[idx].load();
        while (node) {
            if (node->key == key) {
                value = node->value;
                return true;
            }
            node = node->next.load();
        }
        return false;
    }

    // 删除键值对
    bool erase(const K& key) {
        size_t idx = hash_(key) % Capacity;
        Node* prev = nullptr;
        // ... 删除逻辑实现
        return true;
    }
};