·55 分钟
C++ 多线程编程实战：从基础到高级
C++多线程并发异步编程C/C++
线程基础
1. 创建线程
cpp
#include <iostream>
#include <thread>
// 普通函数
void printHello() {
std::cout << "Hello from thread!" << std::endl;
}
// 带参数的函数
void printNumber(int n) {
std::cout << "Number: " << n << std::endl;
}
// Lambda 表达式
auto lambda = [](int x) {
std::cout << "Lambda: " << x << std::endl;
};
// 函数对象
struct Functor {
void operator()(int x) const {
std::cout << "Functor: " << x << std::endl;
}
};
int main() {
// 1. 普通函数
std::thread t1(printHello);
// 2. 带参数
std::thread t2(printNumber, 42);
// 3. Lambda
std::thread t3(lambda, 100);
// 4. 函数对象
std::thread t4(Functor(), 200);
// 5. 成员函数
class MyClass {
public:
void func(int x) {
std::cout << "Member function: " << x << std::endl;
}
};
MyClass obj;
std::thread t5(&MyClass::func, &obj, 300);
// 等待所有线程完成
t1.join();
t2.join();
t3.join();
t4.join();
t5.join();
return 0;
}
2. 线程管理
cpp
#include <thread>
#include <chrono>
void longTask() {
std::this_thread::sleep_for(std::chrono::seconds(2));
std::cout << "Task completed" << std::endl;
}
int main() {
std::thread t(longTask);
// 检查线程是否可 join
if (t.joinable()) {
std::cout << "Thread is joinable" << std::endl;
}
// 获取线程 ID
std::cout << "Thread ID: " << t.get_id() << std::endl;
// 获取硬件并发数
unsigned int cores = std::thread::hardware_concurrency();
std::cout << "Hardware concurrency: " << cores << std::endl;
// 分离线程（后台运行）
t.detach();
// 等待一下，让分离的线程完成
std::this_thread::sleep_for(std::chrono::seconds(3));
return 0;
}
3. 线程局部存储
cpp
#include <thread>
#include <iostream>
// 线程局部变量
thread_local int threadId = 0;
void worker(int id) {
threadId = id; // 每个线程有自己的副本
std::cout << "Thread " << threadId << " started" << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Thread " << threadId << " finished" << std::endl;
}
int main() {
std::thread t1(worker, 1);
std::thread t2(worker, 2);
std::thread t3(worker, 3);
t1.join();
t2.join();
t3.join();
// 主线程的 threadId 仍然是 0
std::cout << "Main thread ID: " << threadId << std::endl;
return 0;
}
同步机制
1. 互斥锁（Mutex）
cpp
#include <mutex>
#include <thread>
#include <vector>
class Counter {
private:
int count = 0;
std::mutex mtx;
public:
void increment() {
std::lock_guard<std::mutex> lock(mtx);
count++;
}
int getCount() const {
return count;
}
};
void worker(Counter& counter, int iterations) {
for (int i = 0; i < iterations; i++) {
counter.increment();
}
}
int main() {
Counter counter;
std::vector<std::thread> threads;
// 创建 10 个线程
for (int i = 0; i < 10; i++) {
threads.emplace_back(worker, std::ref(counter), 1000);
}
// 等待所有线程完成
for (auto& t : threads) {
t.join();
}
std::cout << "Final count: " << counter.getCount() << std::endl;
// 输出：Final count: 10000
return 0;
}
2. 锁的类型
cpp
#include <mutex>
#include <shared_mutex>
std::mutex mtx;
std::recursive_mutex rmtx;
std::timed_mutex tmtx;
std::shared_mutex smtx;
// 1. std::lock_guard - RAII 风格
void func1() {
std::lock_guard<std::mutex> lock(mtx);
// 自动解锁
}
// 2. std::unique_lock - 更灵活
void func2() {
std::unique_lock<std::mutex> lock(mtx, std::defer_lock);
// 延迟加锁
lock.lock(); // 手动加锁
// ...
lock.unlock(); // 手动解锁
lock.lock(); // 可以再次加锁
}
// 3. std::shared_lock - 读写锁
void reader() {
std::shared_lock<std::shared_mutex> lock(smtx);
// 多个线程可以同时读
}
void writer() {
std::unique_lock<std::shared_mutex> lock(smtx);
// 独占写
}
// 4. std::scoped_lock (C++17) - 同时锁定多个互斥量
void transfer(int& from, int& to, int amount) {
std::scoped_lock lock(mtx1, mtx2); // 同时锁定
from -= amount;
to += amount;
}
3. 条件变量
cpp
#include <condition_variable>
#include <queue>
#include <thread>
template<typename T>
class ThreadSafeQueue {
private:
std::queue<T> queue;
mutable std::mutex mtx;
std::condition_variable cv;
public:
void push(T value) {
std::lock_guard<std::mutex> lock(mtx);
queue.push(std::move(value));
cv.notify_one();
}
T pop() {
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock, [this] { return !queue.empty(); });
T value = std::move(queue.front());
queue.pop();
return value;
}
bool tryPop(T& value) {
std::lock_guard<std::mutex> lock(mtx);
if (queue.empty()) return false;
value = std::move(queue.front());
queue.pop();
return true;
}
size_t size() const {
std::lock_guard<std::mutex> lock(mtx);
return queue.size();
}
};
// 生产者-消费者模式
void producer(ThreadSafeQueue<int>& queue) {
for (int i = 0; i < 10; i++) {
queue.push(i);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
void consumer(ThreadSafeQueue<int>& queue) {
for (int i = 0; i < 10; i++) {
int value = queue.pop();
std::cout << "Consumed: " << value << std::endl;
}
}
int main() {
ThreadSafeQueue<int> queue;
std::thread t1(producer, std::ref(queue));
std::thread t2(consumer, std::ref(queue));
t1.join();
t2.join();
return 0;
}
4. 信号量（C++20）
cpp
#include <semaphore>
#include <thread>
#include <vector>
// C++20 信号量
std::counting_semaphore<10> semaphore(3); // 最多 3 个并发
void worker(int id) {
semaphore.acquire(); // 获取许可
std::cout << "Worker " << id << " started" << std::endl;
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Worker " << id << " finished" << std::endl;
semaphore.release(); // 释放许可
}
int main() {
std::vector<std::thread> threads;
for (int i = 0; i < 10; i++) {
threads.emplace_back(worker, i);
}
for (auto& t : threads) {
t.join();
}
return 0;
}
原子操作
1. 原子变量
cpp
#include <atomic>
#include <thread>
#include <vector>
std::atomic<int> atomicCount(0);
int normalCount = 0;
void increment(int iterations) {
for (int i = 0; i < iterations; i++) {
atomicCount++; // 原子操作
normalCount++; // 非原子操作
}
}
int main() {
std::vector<std::thread> threads;
for (int i = 0; i < 10; i++) {
threads.emplace_back(increment, 10000);
}
for (auto& t : threads) {
t.join();
}
std::cout << "Atomic count: " << atomicCount << std::endl; // 总是 100000
std::cout << "Normal count: " << normalCount << std::endl; // 可能小于 100000
return 0;
}
2. 原子操作类型
cpp
#include <atomic>
std::atomic<int> a(0);
// 基本操作
a.store(10); // 存储
int val = a.load(); // 加载
a.exchange(20); // 交换
// 读-改-写操作
a++; // 原子自增
a--; // 原子自减
a += 5; // 原子加
a -= 3; // 原子减
// 比较并交换（CAS）
int expected = 10;
bool success = a.compare_exchange_strong(expected, 20);
if (success) {
// a 从 10 变为 20
} else {
// expected 被更新为 a 的当前值
}
// 内存顺序
a.store(10, std::memory_order_relaxed); // 宽松
a.store(10, std::memory_order_acquire); // 获取
a.store(10, std::memory_order_release); // 释放
a.store(10, std::memory_order_seq_cst); // 顺序一致（默认）
3. 无锁数据结构
cpp
#include <atomic>
#include <memory>
// 无锁栈
template<typename T>
class LockFreeStack {
private:
struct Node {
T data;
Node* next;
Node(T const& data) : data(data), next(nullptr) {}
};
std::atomic<Node*> head;
public:
LockFreeStack() : head(nullptr) {}
void push(T const& data) {
Node* newNode = new Node(data);
newNode->next = head.load();
while (!head.compare_exchange_weak(
newNode->next, newNode)) {
// CAS 失败，重试
}
}
std::shared_ptr<T> pop() {
Node* oldHead = head.load();
while (oldHead && !head.compare_exchange_weak(
oldHead, oldHead->next)) {
// CAS 失败，重试
}
if (oldHead) {
std::shared_ptr<T> result =
std::make_shared<T>(std::move(oldHead->data));
delete oldHead;
return result;
}
return nullptr;
}
~LockFreeStack() {
while (pop()) {}
}
};
异步编程
1. std::async 和 std::future
cpp
#include <future>
#include <iostream>
int compute(int x) {
std::this_thread::sleep_for(std::chrono::seconds(2));
return x * x;
}
int main() {
// 异步执行
std::future<int> result = std::async(std::launch::async, compute, 10);
// 做其他工作
std::cout << "Doing other work..." << std::endl;
// 获取结果（阻塞）
int value = result.get();
std::cout << "Result: " << value << std::endl;
return 0;
}
2. std::promise
cpp
#include <future>
#include <thread>
void producer(std::promise<int> prom) {
std::this_thread::sleep_for(std::chrono::seconds(2));
prom.set_value(42); // 设置结果
}
void consumer(std::future<int> fut) {
std::cout << "Waiting for result..." << std::endl;
int value = fut.get(); // 阻塞等待
std::cout << "Result: " << value << std::endl;
}
int main() {
std::promise<int> prom;
std::future<int> fut = prom.get_future();
std::thread t1(producer, std::move(prom));
std::thread t2(consumer, std::move(fut));
t1.join();
t2.join();
return 0;
}
3. std::packaged_task
cpp
#include <future>
#include <queue>
#include <thread>
template<typename T>
class TaskQueue {
private:
std::queue<std::packaged_task<T()>> tasks;
std::mutex mtx;
public:
void addTask(std::packaged_task<T()> task) {
std::lock_guard<std::mutex> lock(mtx);
tasks.push(std::move(task));
}
std::packaged_task<T()> getTask() {
std::lock_guard<std::mutex> lock(mtx);
if (tasks.empty()) return nullptr;
auto task = std::move(tasks.front());
tasks.pop();
return task;
}
};
void worker(TaskQueue<int>& queue) {
while (true) {
auto task = queue.getTask();
if (!task) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
continue;
}
task(); // 执行任务
}
}
int main() {
TaskQueue<int> queue;
// 添加任务
for (int i = 0; i < 5; i++) {
std::packaged_task<int()> task([i]() {
return i * i;
});
auto future = task.get_future();
queue.addTask(std::move(task));
// 在其他线程获取结果
std::cout << "Result: " << future.get() << std::endl;
}
return 0;
}
4. 协程（C++20）
cpp
#include <coroutine>
#include <iostream>
#include <future>
// 简单的协程任务
template<typename T>
struct Task {
struct promise_type {
T value;
Task get_return_object() {
return Task{
std::coroutine_handle<promise_type>::from_promise(*this)
};
}
std::suspend_never initial_suspend() { return {}; }
std::suspend_never final_suspend() noexcept { return {}; }
void return_value(T v) {
value = v;
}
void unhandled_exception() {
std::terminate();
}
};
std::coroutine_handle<promise_type> handle;
T get() {
return handle.promise().value;
}
};
Task<int> computeAsync(int x) {
co_await std::suspend_always{};
co_return x * x;
}
int main() {
auto task = computeAsync(10);
std::cout << "Result: " << task.get() << std::endl;
return 0;
}
并发模式
1. 线程池
cpp
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
class ThreadPool {
private:
std::vector<std::thread> workers;
std::queue<std::function<void()>> tasks;
std::mutex mtx;
std::condition_variable cv;
bool stop = false;
public:
ThreadPool(size_t numThreads) {
for (size_t i = 0; i < numThreads; i++) {
workers.emplace_back([this] {
while (true) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock, [this] {
return stop || !tasks.empty();
});
if (stop && tasks.empty()) {
return;
}
task = std::move(tasks.front());
tasks.pop();
}
task();
}
});
}
}
~ThreadPool() {
{
std::lock_guard<std::mutex> lock(mtx);
stop = true;
}
cv.notify_all();
for (auto& worker : workers) {
worker.join();
}
}
template<typename F, typename... Args>
auto 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> result = task->get_future();
{
std::lock_guard<std::mutex> lock(mtx);
if (stop) {
throw std::runtime_error("enqueue on stopped ThreadPool");
}
tasks.emplace([task]() { (*task)(); });
}
cv.notify_one();
return result;
}
};
// 使用示例
int main() {
ThreadPool pool(4);
std::vector<std::future<int>> results;
for (int i = 0; i < 8; i++) {
results.emplace_back(
pool.enqueue([i] {
std::this_thread::sleep_for(std::chrono::seconds(1));
return i * i;
})
);
}
for (auto& result : results) {
std::cout << result.get() << " ";
}
std::cout << std::endl;
return 0;
}
2. 读写锁模式
cpp
#include <shared_mutex>
#include <vector>
#include <thread>
class ThreadSafeVector {
private:
std::vector<int> data;
mutable std::shared_mutex mtx;
public:
void push_back(int value) {
std::unique_lock<std::shared_mutex> lock(mtx);
data.push_back(value);
}
int get(size_t index) const {
std::shared_lock<std::shared_mutex> lock(mtx);
return data.at(index);
}
size_t size() const {
std::shared_lock<std::shared_mutex> lock(mtx);
return data.size();
}
void set(size_t index, int value) {
std::unique_lock<std::shared_mutex> lock(mtx);
data.at(index) = value;
}
};
// 多读者-单写者模式
void reader(const ThreadSafeVector& vec, int id) {
for (int i = 0; i < 100; i++) {
int value = vec.get(i % vec.size());
// 处理数据
}
}
void writer(ThreadSafeVector& vec, int id) {
for (int i = 0; i < 10; i++) {
vec.push_back(id * 100 + i);
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
性能优化
1. 避免伪共享
cpp
#include <thread>
#include <vector>
// 伪共享问题
struct BadCounter {
int count1;
int count2; // 可能在同一缓存行
};
// 解决方案：填充
struct GoodCounter {
alignas(64) int count1; // 64 字节对齐
alignas(64) int count2;
};
// 或者使用 C++17 硬件干扰大小
struct Counter {
alignas(std::hardware_destructive_interference_size) int count1;
alignas(std::hardware_destructive_interference_size) int count2;
};
2. 无锁编程技巧
cpp
#include <atomic>
// 使用原子操作代替锁
std::atomic_flag lock = ATOMIC_FLAG_INIT;
void acquire() {
while (lock.test_and_set(std::memory_order_acquire)) {
// 自旋等待
}
}
void release() {
lock.clear(std::memory_order_release);
}
// 使用原子计数器
std::atomic<size_t> counter(0);
size_t getNextId() {
return counter.fetch_add(1, std::memory_order_relaxed);
}
常见问题与解决方案
1. 竞态条件
cpp
// 问题
int shared = 0;
void bad() {
shared++; // 非原子操作
}
// 解决方案
std::atomic<int> safe_shared(0);
void good() {
safe_shared++; // 原子操作
}
2. 死锁
cpp
// 问题
std::mutex mtx1, mtx2;
void thread1() {
std::lock_guard<std::mutex> lock1(mtx1);
std::lock_guard<std::mutex> lock2(mtx2);
}
void thread2() {
std::lock_guard<std::mutex> lock2(mtx2);
std::lock_guard<std::mutex> lock1(mtx1);
}
// 解决方案
void safe_thread1() {
std::scoped_lock lock(mtx1, mtx2); // C++17
}
// 或者固定顺序
void safe_thread2() {
std::lock_guard<std::mutex> lock1(mtx1);
std::lock_guard<std::mutex> lock2(mtx2);
}
3. 线程饥饿
cpp
// 使用公平锁
class FairLock {
private:
std::mutex mtx;
std::condition_variable cv;
int serving = 0;
int ticket = 0;
public:
void lock() {
std::unique_lock<std::mutex> lock(mtx);
int myTicket = ticket++;
cv.wait(lock, [&] { return myTicket == serving; });
}
void unlock() {
std::lock_guard<std::mutex> lock(mtx);
serving++;
cv.notify_all();
}
};
最佳实践总结
设计原则
- 最小化共享：减少线程间共享的数据
- 不可变性：尽量使用 const 和不可变对象
- RAII：使用 RAII 管理锁和资源
- 高层抽象：优先使用线程池、future 等高层抽象
性能考虑
- 避免过度同步：只在必要时加锁
- 使用原子操作：对于简单操作，原子操作比锁更高效
- 考虑无锁结构：对于高性能场景
- 注意缓存行：避免伪共享
调试技巧
- ThreadSanitizer：检测数据竞争
- 死锁检测：使用 std::lock 避免死锁
- 日志记录：记录线程操作
- 单元测试：多线程代码的测试
总结
核心概念
- 线程管理：创建、同步、销毁
- 同步机制：互斥锁、条件变量、原子操作
- 异步编程：future、promise、协程
- 并发模式：线程池、生产者-消费者、读写锁
常见问题
- 竞态条件
- 死锁
- 线程饥饿
- 伪共享
学习资源
延伸阅读：