当前位置：首页 > news >正文

C++多线程与并发中线程池设计、锁优化

news 来源：原创 2025/6/16 12:54:14

一、线程池设计

概念：线程池是一种管理线程资源的机制，预先创建一定数量的线程，当有任务提交时，从线程池中获取线程执行任务，任务完成后线程不会销毁而是返回线程池等待下一个任务。
核心组件
- 线程管理器：负责创建、销毁和管理线程。
- 任务队列：存储待执行的任务。
- 任务接口：定义任务的执行方式。
C++实现示例

#include <vector>
#include <queue>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
#include <stdexcept>class ThreadPool {
public:ThreadPool(size_t threads);template<class F, class... Args>auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type>;~ThreadPool();
private:// 工作线程集合std::vector<std::thread> workers;// 任务队列std::queue<std::function<void()>> tasks;// 同步相关std::mutex queue_mutex;std::condition_variable condition;bool stop;
};// 构造函数，创建指定数量的工作线程
inline ThreadPool::ThreadPool(size_t threads): stop(false) {for(size_t i = 0; i < threads; ++i) {workers.emplace_back([this] {while(true) {std::function<void()> task;{std::unique_lock<std::mutex> lock(this->queue_mutex);this->condition.wait(lock, [this] { return this->stop || !this->tasks.empty(); });if(this->stop && this->tasks.empty())return;task = std::move(this->tasks.front());this->tasks.pop();}task();}});}
}// 添加新的任务到线程池
template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> {using return_type = typename std::result_of<F(Args...)>::type;auto task = std::make_shared<std::packaged_task<return_type()>>(std::bind(std::forward<F>(f), std::forward<Args>(args)...));std::future<return_type> res = task->get_future();{std::unique_lock<std::mutex> lock(queue_mutex);// 不允许向已停止的线程池添加任务if(stop)throw std::runtime_error("enqueue on stopped ThreadPool");tasks.emplace([task]() { (*task)(); });}condition.notify_one();return res;
}// 析构函数，销毁线程池
inline ThreadPool::~ThreadPool() {{std::unique_lock<std::mutex> lock(queue_mutex);stop = true;}condition.notify_all();for(std::thread &worker : workers) {worker.join();}
}

使用示例

// 创建一个包含4个线程的线程池
ThreadPool pool(4);// 向线程池提交任务
std::vector<std::future<int>> results;for(int i = 0; i < 8; ++i) {results.emplace_back(pool.enqueue([i] {std::cout << "hello " << i << std::endl;std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "world " << i << std::endl;return i*i;}));
}// 等待所有任务完成并获取结果
for(auto && result : results)std::cout << result.get() << ' ';
std::cout << std::endl;

二、锁优化

读写锁

概念：读写锁允许多个线程同时读取共享资源，但在写操作时会独占资源，实现“多读单写”的机制。
C++实现

#include <shared_mutex>  // C++17引入
#include <mutex>
#include <map>
#include <string>class ThreadSafeHashMap {
private:mutable std::shared_mutex mtx;std::map<std::string, int> data;public:// 读操作，允许多个线程同时进行int get(const std::string& key) const {std::shared_lock<std::shared_mutex> lock(mtx);auto it = data.find(key);return it != data.end() ? it->second : -1;}// 写操作，独占锁void set(const std::string& key, int value) {std::unique_lock<std::shared_mutex> lock(mtx);data[key] = value;}
};

适用场景：适用于读多写少的场景，如缓存系统。

无锁队列

概念：无锁队列使用原子操作替代传统的锁机制，避免线程阻塞，提高并发性能。
C++实现（基于CAS操作）

#include <atomic>
#include <memory>template<typename T>
class LockFreeQueue {
private:struct Node {T data;std::atomic<Node*> next;Node(const T& value) : data(value), next(nullptr) {}};std::atomic<Node*> head;std::atomic<Node*> tail;public:LockFreeQueue() : head(nullptr), tail(nullptr) {}void enqueue(const T& value) {Node* newNode = new Node(value);Node* oldTail = tail.load();while (!tail.compare_exchange_weak(oldTail, newNode)) {}if (oldTail) {oldTail->next = newNode;} else {head = newNode;}}bool dequeue(T& value) {Node* oldHead = head.load();while (oldHead) {Node* next = oldHead->next.load();if (head.compare_exchange_weak(oldHead, next)) {value = oldHead->data;delete oldHead;return true;}}return false;}~LockFreeQueue() {T value;while (dequeue(value)) {}}
};

适用场景：适用于高性能要求的生产者-消费者模型。

性能对比与选择策略

技术	优点	缺点	适用场景
传统互斥锁	实现简单，适用广泛	线程阻塞，上下文切换开销大	竞争不激烈的场景
读写锁	提高读并发性能	实现复杂，写操作可能饥饿	读多写少的场景
无锁队列	无阻塞，高性能	实现复杂，调试困难	高性能要求的队列操作