C++与现代开发实践第三节：多线程与并发编程

第四章：C++与现代开发实践

第三节：多线程与并发编程

在这一课中，我们将详细探讨多线程与并发编程的各个方面，特别是从线程的创建、管理到高级的优化技术，并且通过复杂的实战案例来展示如何应对并发问题。最后，我们会通过一个项目实践展示如何将这些理论应用到实际开发中。

1. 线程的创建、管理与优化

多线程编程中的核心之一是如何有效地创建和管理线程，并对其性能进行优化。在C++中，C++11标准引入了原生的线程支持，使得多线程编程更加直接和高效。

1.1 线程的创建

C++11提供了 std::thread 类来实现多线程编程，创建线程非常方便，但在实际开发中，我们经常遇到比简单的函数调用更复杂的需求。

1.1.1 创建带参数的线程

我们可以通过 std::thread 的构造函数直接传递参数，这在实际开发中非常有用。例如，当我们需要启动多个任务并为每个任务传递不同的参数时，这种方式能够简化代码。

#include <iostream>
#include <thread>
#include <vector>void task(int id, const std::string& message) {std::cout << "Task " << id << ": " << message << std::endl;
}int main() {std::vector<std::thread> threads;for (int i = 0; i < 5; ++i) {threads.emplace_back(task, i, "Hello from thread");}for (auto& th : threads) {th.join();}return 0;
}

1.1.2 使用 `std::bind` 和 lambda 表达式创建线程

C++11 引入的 std::bind 和 lambda 表达式可以极大提高代码的灵活性。当我们需要传递额外的上下文或处理一些复杂逻辑时，这些功能尤其有用。

#include <iostream>
#include <thread>
#include <functional>
#include <string>void printMessage(const std::string& message, int count) {for (int i = 0; i < count; ++i) {std::cout << message << std::endl;}
}int main() {std::string msg = "Hello from lambda!";int repeat = 3;std::thread t1(std::bind(printMessage, msg, repeat));std::thread t2([msg, repeat]() {for (int i = 0; i < repeat; ++i) {std::cout << msg << std::endl;}});t1.join();t2.join();return 0;
}

在这个示例中，我们通过 std::bind 绑定了函数参数，同时使用了 lambda 表达式直接定义了一个线程内的任务。

1.1.3 使用类成员函数作为线程入口

当需要在类中使用线程时，直接调用类的成员函数是常见的做法。然而，由于成员函数有一个隐式的 this 指针，因此需要特别处理。在 C++11 中，我们可以使用 std::bind 或 lambda 表达式来传递成员函数。

#include <iostream>
#include <thread>class Worker {
public:void doWork(int id) {std::cout << "Worker " << id << " is working..." << std::endl;}
};int main() {Worker worker;std::thread t(&Worker::doWork, &worker, 1); // 使用成员函数t.join();return 0;
}

在这个例子中，std::thread 能够直接调用类的成员函数，注意我们需要传递 this 指针来调用非静态成员函数。

1.2 线程管理

线程的管理包括如何合理地创建、销毁线程以及如何协调线程之间的工作。以下是一些常见的线程管理技术。

1.2.1 `join()` 和 `detach()`

join()：调用 join() 会阻塞当前线程，直到目标线程完成为止。这是最常见的等待线程完成的方法。
detach()：当线程被分离时，它将在后台独立运行，不再与主线程同步。如果线程对象在没有调用 join() 或 detach() 的情况下销毁，会导致程序崩溃。

#include <iostream>
#include <thread>void task() {std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "Task done!" << std::endl;
}int main() {std::thread t(task);t.detach(); // 分离线程std::this_thread::sleep_for(std::chrono::seconds(2)); // 等待任务完成return 0;
}

1.2.2 使用线程池管理线程

在现代开发中，频繁创建和销毁线程会导致性能问题。为了解决这个问题，线程池成为了一种有效的解决方案。线程池通过预先创建固定数量的线程来处理任务，避免了线程创建和销毁的开销。

C++标准库并未提供原生的线程池实现，但我们可以通过 std::thread 和任务队列来实现一个简单的线程池。

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <functional>
#include <condition_variable>class ThreadPool {
public:ThreadPool(size_t numThreads);~ThreadPool();void enqueue(std::function<void()> task);private:std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::mutex queueMutex;std::condition_variable condition;bool stop;
};ThreadPool::ThreadPool(size_t numThreads) : stop(false) {for (size_t i = 0; i < numThreads; ++i) {workers.emplace_back([this] {while (true) {std::function<void()> task;{std::unique_lock<std::mutex> lock(this->queueMutex);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();}});}
}ThreadPool::~ThreadPool() {{std::unique_lock<std::mutex> lock(queueMutex);stop = true;}condition.notify_all();for (std::thread &worker : workers) {worker.join();}
}void ThreadPool::enqueue(std::function<void()> task) {{std::unique_lock<std::mutex> lock(queueMutex);tasks.push(std::move(task));}condition.notify_one();
}int main() {ThreadPool pool(4);for (int i = 0; i < 8; ++i) {pool.enqueue([i] {std::cout << "Task " << i << " is being processed." << std::endl;});}std::this_thread::sleep_for(std::chrono::seconds(2)); // 等待任务完成return 0;
}

这个简单的线程池实现允许我们将任务添加到队列中，并由固定数量的线程来处理它们。

1.3 线程同步

多线程编程中，线程同步是至关重要的。共享资源的并发访问可能会导致数据竞争（race condition）。C++提供了多种同步机制来防止此类问题。

1.3.1 使用 `std::mutex` 进行线程同步

std::mutex 是一种基本的同步原语，用于在多个线程之间保护共享资源。通过加锁机制，确保同一时刻只有一个线程能够访问某个资源。

#include <iostream>
#include <thread>
#include <mutex>std::mutex mtx;void printSafe(const std::string& message) {std::lock_guard<std::mutex> lock(mtx); // 自动加锁和解锁std::cout << message << std::endl;
}int main() {std::thread t1(printSafe, "Thread 1: Safe printing");std::thread t2(printSafe, "Thread 2: Safe printing");t1.join();t2.join();return 0;
}

在上面的例子中，我们使用 std::mutex 来确保同一时间只有一个线程能够执行 printSafe 函数，防止输出混乱。

1.3.2 动态内存的管理策略与技巧

动态内存管理是 C++ 编程中的一个重要概念，它直接影响程序的性能、稳定性和可维护性。在现代 C++ 中，合理的内存管理策略能够有效避免内存泄漏、悬空指针和其他潜在问题。以下是一些常用的动态内存管理策略和技巧。

1.3.2.1 使用智能指针

智能指针是 C++11 引入的重要特性，主要包括 std::unique_ptr、std::shared_ptr 和 std::weak_ptr。这些智能指针提供了自动的内存管理，有效避免了手动管理内存时可能出现的错误。

1. std::unique_ptr

std::unique_ptr 表示对动态分配对象的独占所有权。它确保在其作用域结束时自动释放内存。示例如下：

#include <iostream>
#include <memory>void uniquePtrExample() {std::unique_ptr<int> ptr = std::make_unique<int>(42);std::cout << "Value: " << *ptr << std::endl;  // 输出：Value: 42// 不需要手动 delete，ptr 超出作用域后会自动释放
}

2. std::shared_ptr

std::shared_ptr 允许多个指针共享同一个对象。当最后一个 shared_ptr 被销毁时，才会释放对象。它适用于多个所有者的场景。

#include <iostream>
#include <memory>void sharedPtrExample() {std::shared_ptr<int> ptr1 = std::make_shared<int>(42);{std::shared_ptr<int> ptr2 = ptr1;  // 共享所有权std::cout << "Value: " << *ptr2 << std::endl;  // 输出：Value: 42}  // ptr2 超出作用域，ptr1 仍然存在std::cout << "Value after ptr2 scope: " << *ptr1 << std::endl;  // 仍然可用
}

3. std::weak_ptr

std::weak_ptr 与 std::shared_ptr 配合使用，解决了循环引用的问题。它不拥有对象的所有权，因此不会影响引用计数。

#include <iostream>
#include <memory>class Node {
public:std::shared_ptr<Node> next;~Node() { std::cout << "Node destroyed\n"; }
};void weakPtrExample() {std::shared_ptr<Node> node1 = std::make_shared<Node>();std::weak_ptr<Node> weakNode = node1;  // 不增加引用计数std::cout << "Use count: " << weakNode.use_count() << std::endl;  // 输出：1
}

1.3.2.2 内存池的使用

内存池是一种高效的内存管理策略，特别适用于频繁分配和释放小块内存的场景。通过预先分配一块大内存，内存池能够减少内存分配和释放的开销。

#include <iostream>
#include <vector>class MemoryPool {
public:MemoryPool(size_t blockSize, size_t blockCount): blockSize(blockSize), pool(blockCount * blockSize) {for (size_t i = 0; i < blockCount; ++i) {freeBlocks.push_back(&pool[i * blockSize]);}}void* allocate() {if (freeBlocks.empty()) return nullptr;void* block = freeBlocks.back();freeBlocks.pop_back();return block;}void deallocate(void* block) {freeBlocks.push_back(block);}private:size_t blockSize;std::vector<char> pool;std::vector<void*> freeBlocks;
};

1.3.2.3 使用内存监控工具

动态内存管理的另一个重要方面是内存监控。使用工具（如 Valgrind、AddressSanitizer 和 Visual Studio 的内存分析器）能够帮助开发者检测内存泄漏、越界访问和其他内存相关问题。

Valgrind：这是一个强大的工具，用于检测内存泄漏和错误。
```
valgrind --leak-check=full ./your_program
```
AddressSanitizer：这是一个编译器提供的工具，能够在运行时检测内存错误。只需在编译时添加 -fsanitize=address。

1.3.2.4 避免不必要的动态内存分配

在设计程序时，应尽量减少动态内存分配，尤其是在性能敏感的代码路径中。可以考虑以下策略：

使用栈分配：如果对象的生命周期相对短暂且大小可预见，尽量使用栈分配。
预分配内存：对于重复使用的对象，考虑在程序启动时预分配所需内存。

1.3.2.5 编写 RAII 风格的类

RAII（资源获取即初始化）是一种常用的内存管理技术，它通过类的构造和析构函数自动管理资源。编写 RAII 风格的类，能够确保资源在不再需要时自动释放。

#include <iostream>
#include <fstream>class FileHandler {
public:FileHandler(const std::string& filename) : file(filename) {if (!file.is_open()) {throw std::runtime_error("Failed to open file");}}~FileHandler() {if (file.is_open()) {file.close();}}private:std::ofstream file;
};

通过上述策略和技巧，开发者能够更有效地管理动态内存，减少内存相关错误，提高程序的性能和稳定性。

1.3.3 使用条件变量 `std::condition_variable`

条件变量是一种更高级的线程同步工具，能够让一个或多个线程等待某个条件的满足。通过条件变量，可以在不忙等（busy waiting）的情况下，使线程在某些条件下等待并被通知。

条件变量通常与 std::mutex 结合使用：线程等待某个条件（如队列不为空），当条件满足时，通知等待线程继续执行。

示例：生产者-消费者模式

生产者-消费者模式是多线程编程中常见的一种应用场景。在这种模式中，生产者线程向队列中添加数据，消费者线程从队列中取出数据进行处理。

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>std::queue<int> buffer;
std::mutex mtx;
std::condition_variable cv;
bool finished = false;void producer(int items) {for (int i = 0; i < items; ++i) {std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟生产std::unique_lock<std::mutex> lock(mtx);buffer.push(i);std::cout << "Produced: " << i << std::endl;cv.notify_one();  // 通知消费者有新的数据}std::unique_lock<std::mutex> lock(mtx);finished = true;cv.notify_all();  // 通知所有消费者生产结束
}void consumer() {while (true) {std::unique_lock<std::mutex> lock(mtx);cv.wait(lock, [] { return !buffer.empty() || finished; });while (!buffer.empty()) {int item = buffer.front();buffer.pop();std::cout << "Consumed: " << item << std::endl;}if (finished && buffer.empty()) break;  // 如果生产者结束并且队列为空，则退出}
}int main() {std::thread prod1(producer, 10);std::thread cons1(consumer), cons2(consumer);prod1.join();cons1.join();cons2.join();return 0;
}

在这个例子中，生产者生产一定数量的任务，放入队列中。消费者线程从队列中获取任务并处理。通过 std::condition_variable 实现了生产者与消费者之间的高效同步。cv.notify_one() 和 cv.notify_all() 分别用于通知一个或所有等待线程。

1.4 线程优化技巧

优化多线程程序是并发编程中最具有挑战性的任务之一。优化不仅包括避免死锁和数据竞争，还涉及到如何高效利用多核处理器。

1.4.1 减少锁的使用

锁的过度使用会导致性能下降，因为线程必须等待锁释放。为了提高性能，我们应该尽可能减少锁的使用，甚至可以通过无锁编程来优化部分代码。

分片加锁：将大块的代码分成更小的片段，并在每个片段上使用锁，从而减少锁的持有时间。
局部锁：如果锁只需要保护某个局部的数据，可以将锁的作用范围限制到局部变量而非整个函数。

1.4.2 使用 `std::shared_mutex` 实现读写锁

std::shared_mutex 是 C++17 引入的一种互斥锁，它允许多个线程同时读，但只有一个线程能够写。当多线程程序中读操作远多于写操作时，读写锁能够显著提高性能。

#include <iostream>
#include <shared_mutex>
#include <thread>
#include <vector>std::shared_mutex rw_mutex;
int shared_data = 0;void reader(int id) {std::shared_lock lock(rw_mutex);  // 共享锁，多个读者可以同时持有std::cout << "Reader " << id << ": Read " << shared_data << std::endl;
}void writer(int id, int value) {std::unique_lock lock(rw_mutex);  // 独占锁，写者独占资源shared_data = value;std::cout << "Writer " << id << ": Wrote " << shared_data << std::endl;
}int main() {std::vector<std::thread> threads;// 创建5个读者和2个写者for (int i = 0; i < 5; ++i) {threads.emplace_back(reader, i);}for (int i = 0; i < 2; ++i) {threads.emplace_back(writer, i, i * 10);}for (auto& th : threads) {th.join();}return 0;
}

在这个例子中，std::shared_mutex 允许多个读线程同时访问共享数据，而写线程需要独占该资源。这样可以显著提高读密集型操作的性能。

1.4.3 使用无锁编程（lock-free programming）

无锁编程是一种高级的并发编程技术，旨在通过避免锁来提高性能。无锁编程依赖于原子操作，C++11 提供了 std::atomic 模板类来实现原子操作。

无锁编程适合在高并发环境下使用，能够减少由于锁竞争引起的性能瓶颈。

#include <iostream>
#include <atomic>
#include <thread>
#include <vector>std::atomic<int> counter(0);void increment(int id) {for (int i = 0; i < 10000; ++i) {counter.fetch_add(1, std::memory_order_relaxed);  // 原子操作}
}int main() {std::vector<std::thread> threads;for (int i = 0; i < 10; ++i) {threads.emplace_back(increment, i);}for (auto& th : threads) {th.join();}std::cout << "Final counter value: " << counter << std::endl;return 0;
}

这个例子展示了如何使用 std::atomic 来实现无锁递增操作，避免了传统的互斥锁带来的性能开销。 std::atomic 保证了每个线程都能正确地更新共享变量 counter，从而避免数据竞争。

2. 实战案例分析：解决并发问题

并发编程中，实际遇到的问题往往复杂且多样化。下面我们将通过一个实际的案例，展示如何在高并发环境下解决典型的并发问题。

2.1 案例：高性能日志系统

在大规模服务器应用中，日志记录通常是一个高并发的场景，多个线程同时向日志文件写入数据。如果处理不当，日志系统会成为性能瓶颈，甚至导致数据丢失或写入顺序错误。

需求分析

高并发：多个线程同时写入日志。
线程安全：确保日志的写入顺序不会错乱。
性能优化：避免频繁锁定文件资源，提高日志记录的性能。

实现思路

我们可以使用日志队列和专用的日志线程来实现异步日志系统。所有线程将日志写入到共享队列中，日志线程负责从队列中取出日志并写入文件。这种方式能够减少线程之间的锁竞争，提高性能。

代码实现

#include <iostream>
#include <fstream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <string>
#include <atomic>class AsyncLogger {
public:AsyncLogger() : stopLogging(false) {loggingThread = std::thread(&AsyncLogger::logThread, this);}~AsyncLogger() {{std::unique_lock<std::mutex> lock(queueMutex);stopLogging = true;cv.notify_all();}loggingThread.join();}void log(const std::string& message) {std::unique_lock<std::mutex> lock(queueMutex);logQueue.push(message);cv.notify_one();}private:std::queue<std::string> logQueue;std::mutex queueMutex;std::condition_variable cv;std::atomic<bool> stopLogging;std::thread loggingThread;void logThread() {std::ofstream logFile("log.txt", std::ios::app);while (!stopLogging || !logQueue.empty()) {std::unique_lock<std::mutex> lock(queueMutex);cv.wait(lock, [this] {return !logQueue.empty() || stopLogging;});while (!logQueue.empty()) {logFile << logQueue.front() << std::endl;logQueue.pop();}}logFile.close();}
};void logMessages(AsyncLogger& logger, int id) {for (int i = 0; i < 100; ++i) {logger.log("Thread " + std::to_string(id) + ": Log entry " + std::to_string(i));}
}int main() {AsyncLogger logger;std::thread t1(logMessages, std::ref(logger), 1);std::thread t2(logMessages, std::ref(logger), 2);t1.join();t2.join();return 0;
}

2.2 优化并发日志系统

在上述日志系统的实现中，我们通过将日志写入操作异步化，减少了对主线程的阻塞。然而，仍然存在一些可以优化的方面，以进一步提升性能和可靠性。

2.2.1 减少锁的持有时间

在 AsyncLogger 中，std::unique_lock 被用于保护共享日志队列。在实际使用中，锁的持有时间应该尽量缩短，确保在获取锁后尽快完成队列操作。可以通过减少锁的作用范围和使用局部变量来优化。

void logThread() {std::ofstream logFile("log.txt", std::ios::app);std::queue<std::string> localQueue;  // 本地队列，用于减少锁持有时间while (!stopLogging || !logQueue.empty()) {{std::unique_lock<std::mutex> lock(queueMutex);cv.wait(lock, [this] {return !logQueue.empty() || stopLogging;});// 移动队列内容到本地队列localQueue.swap(logQueue);}// 在不持有锁的情况下处理本地队列while (!localQueue.empty()) {logFile << localQueue.front() << std::endl;localQueue.pop();}}logFile.close();
}

这种方式可以在持有锁的时间内完成更少的工作，从而减少对其他线程的影响。

2.2.2 使用环形缓冲区（Ring Buffer）

环形缓冲区是一种常见的高效数据结构，适合用于生产者-消费者问题。它提供了一个固定大小的数组，允许生产者和消费者同时操作而不需要加锁。这种方式对于高频写入的场景尤其有效。

实现一个简单的环形缓冲区如下：

#include <iostream>
#include <atomic>
#include <thread>
#include <vector>template<typename T>
class RingBuffer {
public:RingBuffer(size_t size) : buffer(size), head(0), tail(0) {}bool push(const T& item) {size_t next = (head + 1) % buffer.size();if (next == tail) return false;  // Buffer is fullbuffer[head] = item;head = next;return true;}bool pop(T& item) {if (tail == head) return false;  // Buffer is emptyitem = buffer[tail];tail = (tail + 1) % buffer.size();return true;}private:std::vector<T> buffer;std::atomic<size_t> head, tail;
};void logThread(RingBuffer<std::string>& logBuffer) {std::ofstream logFile("log.txt", std::ios::app);std::string message;while (true) {if (logBuffer.pop(message)) {logFile << message << std::endl;}}
}int main() {RingBuffer<std::string> logBuffer(1024);std::thread logger(logThread, std::ref(logBuffer));// Logging simulationfor (int i = 0; i < 10000; ++i) {logBuffer.push("Log entry " + std::to_string(i));}logger.join();return 0;
}

通过使用环形缓冲区，我们能够显著减少锁的使用，提高日志记录的性能。

2.3 高级技巧：使用 `std::future` 和 `std::async`

std::future 和 std::async 提供了一种简化的方式来处理异步任务。它们可以方便地获取线程的结果，并支持将任务提交给线程池，从而提高并发性能。

示例：使用 `std::async` 实现异步日志记录

#include <iostream>
#include <future>
#include <thread>
#include <queue>
#include <mutex>std::mutex mtx;
std::queue<std::string> logQueue;void asyncLog(const std::string& message) {std::unique_lock<std::mutex> lock(mtx);logQueue.push(message);std::cout << "Async Log: " << message << std::endl;
}int main() {std::vector<std::future<void>> futures;for (int i = 0; i < 10; ++i) {std::string message = "Log entry " + std::to_string(i);futures.push_back(std::async(std::launch::async, asyncLog, message));}// 等待所有日志任务完成for (auto& future : futures) {future.get();}return 0;
}

在这个示例中，std::async 被用来并发地提交日志记录任务，而主线程可以继续执行其他操作。最终，通过 future.get() 确保所有异步任务完成。

3. 项目实践：构建一个多线程应用

通过上述理论和案例分析，我们已经了解了如何实现高效的多线程程序。下面，我们将把这些知识应用于一个实际的多线程应用项目。

3.1 项目概述

本项目将构建一个简单的 HTTP 服务器，能够处理并发的客户端请求。在这个服务器中，我们将使用线程池来管理工作线程，并使用 std::condition_variable 来协调请求的处理。

3.2 项目结构

HTTPServer：主服务器类，负责管理线程池和监听端口。
ThreadPool：线程池类，用于管理多个工作线程。
RequestHandler：处理 HTTP 请求的类，负责解析请求并生成响应。

3.3 代码实现

首先实现线程池：

#include <iostream>
#include <vector>
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>class ThreadPool {
public:ThreadPool(size_t threads);template<class F>void enqueue(F&& f);~ThreadPool();private:std::vector<std::thread> workers;std::queue<std::function<void()>> tasks;std::mutex queueMutex;std::condition_variable condition;bool stop;
};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->queueMutex);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>
void ThreadPool::enqueue(F&& f) {{std::unique_lock<std::mutex> lock(queueMutex);tasks.emplace(std::forward<F>(f));}condition.notify_one();
}ThreadPool::~ThreadPool() {{std::unique_lock<std::mutex> lock(queueMutex);stop = true;}condition.notify_all();for (std::thread& worker : workers) {worker.join();}
}

接下来，构建 HTTP 服务器：

#include <iostream>
#include <thread>
#include <boost/asio.hpp>using boost::asio::ip::tcp;class HTTPServer {
public:HTTPServer(boost::asio::io_context& io_context, short port);void start();private:void handleRequest(tcp::socket socket);ThreadPool pool;tcp::acceptor acceptor;
};HTTPServer::HTTPServer(boost::asio::io_context& io_context, short port): acceptor(io_context, tcp::endpoint(tcp::v4(), port)), pool(4) {}void HTTPServer::start() {while (true) {tcp::socket socket(acceptor.get_io_context());acceptor.accept(socket);pool.enqueue([this, socket = std::move(socket)]() mutable {handleRequest(std::move(socket));});}
}void HTTPServer::handleRequest(tcp::socket socket) {// 处理 HTTP 请求// 发送简单的响应const std::string response = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, World!";boost::asio::write(socket, boost::asio::buffer(response));
}int main() {try {boost::asio::io_context io_context;HTTPServer server(io_context, 8080);server.start();} catch (std::exception& e) {std::cerr << "Exception: " << e.what() << "\n";}return 0;
}