并发操纵的同步_第四章_《C++并发编程实战》条记

1. 期待变乱或期待其他条件

std::condition_variable：是 C++ 尺度库提供的用于线程间同步的工具，它允许一个或多个线程期待某个条件满足后再继承实验。通常与 std::mutex 一起使用，std::mutex 用于掩护共享数据，std::condition_variable 用于线程间的关照机制。
虚伪唤醒 ：指线程在没有收到明白关照的情况下从 wait 状态中被唤醒。为克制虚伪唤醒，wait 函数通常会担当一个谓词（可调用对象）作为参数，线程被唤醒后会查抄该谓词的返回值，假如为 false 则会继承期待。

1. #include <iostream>
2. #include <thread>
3. #include <mutex>
4. #include <condition_variable>
6. std::mutex mtx;
7. std::condition_variable cv;
8. bool ready = false;
10. void worker() {
11.     std::unique_lock<std::mutex> lock(mtx);
12.     // 等待 ready 变为 true
13.     cv.wait(lock, [] { return ready; });
14.     std::cout << "Worker thread is working." << std::endl;
15. }
17. void prepare_work() {
18.     {
19.         std::lock_guard<std::mutex> lock(mtx);
20.         ready = true;
21.     }
22.     // 通知一个等待的线程
23.     cv.notify_one();
24.     std::cout << "Work is prepared." << std::endl;
25. }
27. int main() {
28.     std::thread t1(worker);
29.     std::thread t2(prepare_work);
31.     t1.join();
32.     t2.join();
34.     return 0;
35. }
36. /*
37. 1. worker 线程首先获取 std::unique_lock 锁定互斥量 mtx，然后调用 cv.wait(lock, [] { return ready; })。wait 函数会检查 ready 条件，如果不满足，它会释放锁并使线程进入等待状态。
38. 2. prepare_work 线程获取锁，将 ready 置为 true，释放锁后调用 cv.notify_one() 通知一个等待的线程。
39. 3. 当 worker 线程被通知后，它会重新获取锁并继续执行。
40. */

复制代码

2. 基于条件变量的生产者 - 斲丧者模子

• 焦点概念
  生产者 - 斲丧者模子是一种常见的并发编程模式，此中生产者线程负责天生数据并将其放入共享缓冲区，斲丧者线程负责从缓冲区中取出数据并举行处置处罚。使用条件变量可以实现生产者和斲丧者之间的同步，克制数据竞争和缓冲区溢出。
  
• 代码
  

1. #include <iostream>
2. #include <queue>
3. #include <thread>
4. #include <mutex>
5. #include <condition_variable>
7. std::queue<int> data_queue;
8. std::mutex mtx;
9. std::condition_variable data_cond;
11. void producer() {
12.     for (int i = 0; i < 5; ++i) {
13.         {
14.             std::lock_guard<std::mutex> lock(mtx);
15.             data_queue.push(i);
16.         }
17.         // 通知消费者有新数据
18.         data_cond.notify_one();
19.         std::this_thread::sleep_for(std::chrono::seconds(1));
20.     }
21. }
23. void consumer() {
24.     while (true) {
25.         std::unique_lock<std::mutex> lock(mtx);
26.         // 等待队列中有数据
27.         data_cond.wait(lock, [] { return!data_queue.empty(); });
28.         int value = data_queue.front();
29.         data_queue.pop();
30.         lock.unlock();
31.         std::cout << "Consumed: " << value << std::endl;
32.     }
33. }
35. int main() {
36.     std::thread t1(producer);
37.     std::thread t2(consumer);
39.     t1.join();
40.     // 这里可以添加适当的机制来停止消费者线程
41.     // 例如使用一个标志位
42.     return 0;
43. }
44. /*
45. 1. producer 线程每隔 1 秒向队列中添加一个数据，并调用 data_cond.notify_one() 通知消费者线程。
46. 2. consumer 线程在 data_cond.wait(lock, [] { return!data_queue.empty(); }) 处等待，直到队列中有数据。当有数据时，它会取出数据并处理。
47. */

复制代码

3. 超时期待

• 焦点概念
  在某些情况下，线程不能无穷期地期待某个条件满足，须要设置一个超时时间。std::condition_variable 提供了 wait_for 和 wait_until 方法来实现超时期待。
  
• 代码
  

1. #include <iostream>
2. #include <thread>
3. #include <mutex>
4. #include <condition_variable>
5. #include <chrono>
7. std::mutex mtx;
8. std::condition_variable cv;
9. bool ready = false;
11. void worker() {
12.     std::unique_lock<std::mutex> lock(mtx);
13.     // 等待 2 秒
14.     auto timeout = std::chrono::steady_clock::now() + std::chrono::seconds(2);
15.     if (cv.wait_until(lock, timeout, [] { return ready; })) {
16.         std::cout << "Worker thread is working after being notified." << std::endl;
17.     } else {
18.         std::cout << "Worker thread timed out." << std::endl;
19.     }
20. }
22. void prepare_work() {
23.     std::this_thread::sleep_for(std::chrono::seconds(3));
24.     {
25.         std::lock_guard<std::mutex> lock(mtx);
26.         ready = true;
27.     }
28.     cv.notify_one();
29.     std::cout << "Work is prepared." << std::endl;
30. }
32. int main() {
33.     std::thread t1(worker);
34.     std::thread t2(prepare_work);
36.     t1.join();
37.     t2.join();
39.     return 0;
40. }
41. /*
42. - worker 线程使用 cv.wait_until(lock, timeout, [] { return ready; }) 等待 ready 条件满足，最多等待 2 秒。
43. - prepare_work 线程在 3 秒后才将 ready 置为 true 并通知 worker 线程，因此 worker 线程会超时。
44. */

复制代码

4. std::future 和 std::promise

焦点概念

std::future：表现一个异步操纵的结果，它可以用来获取异步任务的返回值。std::future 对象可以通过 get() 方法来壅闭期待异步操纵完成并获取结果。
std::promise：是一个可以存储值或非常的对象，它可以与 std::future 关联，用于在差异线程之间转达结果。std::promise 对象可以通过 set_value() 方法设置结果，或通过 set_exception() 方法设置非常。
代码

1. #include <iostream>
2. #include <thread>
3. #include <future>
5. int calculate_sum(int a, int b) {
6.     return a + b;
7. }
9. int main() {
10.     // 创建一个 std::promise 对象
11.     std::promise<int> prom;
12.     // 获取与 promise 关联的 future
13.     std::future<int> fut = prom.get_future();
15.     std::thread t([&prom]() {
16.         int result = calculate_sum(3, 5);
17.         // 设置 promise 的值
18.         prom.set_value(result);
19.     });
21.     // 获取 future 的值
22.     int sum = fut.get();
23.     std::cout << "The sum is: " << sum << std::endl;
25.     t.join();
26.     return 0;
27. }
28. /*
29. - 创建 std::promise<int> prom 和与之关联的 std::future<int> fut。
30. - 启动一个新线程，在该线程中调用 calculate_sum 函数计算结果，并使用 prom.set_value(result) 设置 promise 的值。
31. - 在主线程中，使用 fut.get() 获取异步操作的结果。
32. */

复制代码

5. std::packaged_task 和 std::async

焦点概念

• std::packaged_task: 包装一个可调用对象（如函数、函数对象或 lambda 表达式），并将其结果与 std::future 关联。可以通过调用 std::packaged_task 对象来实验包装的可调用对象，并通过关联的 std::future 获取结果。
  
• std::async:是一个更高级的异步操纵启动方式，它可以主动管理线程和返回结果。std::async 担当一个可调用对象和其参数，返回一个 std::future 对象，用于获取异步操纵的结果。
  
• 示例代码:
  

1. #include <iostream>
2. #include <thread>
3. #include <future>
5. int add(int a, int b) {
6.     return a + b;
7. }
9. int main() {
10.     // 创建一个 packaged_task 对象
11.     std::packaged_task<int(int, int)> task(add);
12.     // 获取与 packaged_task 关联的 future
13.     std::future<int> fut = task.get_future();
15.     std::thread t(std::move(task), 3, 5);
17.     // 获取 future 的值
18.     int result = fut.get();
19.     std::cout << "The result is: " << result << std::endl;
21.     t.join();
22.     return 0;
23. }
24. /*
25. - 创建 std::packaged_task<int(int, int)> task(add) 包装 add 函数，并获取与之关联的 std::future<int> fut。
26. - 启动一个新线程，将 task 移动到线程中执行，并传递参数 3 和 5。
27. - 在主线程中，使用 fut.get() 获取异步操作的结果。
28. */

复制代码

6. 多选标题

• 关于 std::condition_variable，以下说法精确的是（ ）
  A. std::condition_variable 必须与 std::mutex 一起使用
  B. 调用 wait 方法时，假如不使用谓词，大概会发生虚伪唤醒
  C. notify_one 方法会唤醒全部期待的线程
  D. wait_for 和 wait_until 方法可以实现超时期待
  
• 在生产者 - 斲丧者模子中，以下哪些做法可以进步性能和安全性（ ）
  A. 生产者和斲丧者使用同一个互斥量掩护共享队列
  B. 生产者在添加数据后实时关照斲丧者
  C. 斲丧者在处置处罚数据时长时间持有锁
  D. 使用谓词来克制斲丧者的虚伪唤醒
  
• 关于 std::future 和 std::promise，以下说法精确的是（ ）
  A. std::future 可以用来获取异步操纵的结果
  B. std::promise 可以存储值或非常
  C. 一个 std::promise 只能与一个 std::future 关联
  D. std::future 的 get 方法可以多次调用
  
• 以下哪些情况大概导致线程的虚伪唤醒（ ）
  A. 操纵体系的调治战略
  B. 硬件停止
  C. 信号处置处罚
  D. 调用 notify_one 或 notify_all 方法
  
• 在使用 std::condition_variable 时，以下做法精确的是（ ）
  A. 在调用 wait 方法前先锁定互斥量
  B. 在调用 notify_one 或 notify_all 方法时持有互斥量
  C. 使用谓词来克制虚伪唤醒
  D. 在 wait 方法返回后再次查抄条件
  

6. 多选标题答案

• 答案：ABD
  表明：std::condition_variable 依靠 std::mutex 来掩护共享数据，以是必须与 std::mutex 一起使用，A 精确；不使用谓词时，wait 方法大概会因操纵体系或硬件的缘故起因发生虚伪唤醒，B 精确；notify_one 方法只会唤醒一个期待的线程，notify_all 才会唤醒全部期待的线程，C 错误；wait_for 和 wait_until 方法可以让线程在期待肯定时间后主动返回，实现超时期待，D 精确。
  
• 答案：ABD
  表明：生产者和斲丧者使用同一个互斥量掩护共享队列可以克制数据竞争，包管线程安全，A 精确；生产者在添加数据后实时关照斲丧者可以淘汰斲丧者的期待时间，进步性能，B 精确；斲丧者在处置处罚数据时长时间持有锁会导致其他线程无法访问共享队列，低落并发性能，C 错误；使用谓词来克制斲丧者的虚伪唤醒可以确保斲丧者只在队列中有数据时才举行处置处罚，进步步调的精确性和性能，D 精确。
  
• 答案：ABC
  表明：std::future 紧张用于获取异步操纵的结果，A 精确；std::promise 可以存储一个值或非常，并通过关联的 std::future 转达给其他线程，B 精确；一个 std::promise 只能与一个 std::future 关联，C 精确；std::future 的 get 方法只能调用一次，多次调用会导致未界说举动，D 错误。
  
• 答案：ABC
  表明：操纵体系的调治战略、硬件停止和信号处置处罚等因素都大概导致线程的虚伪唤醒，A、B、C 精确；调用 notify_one 或 notify_all 方法是正常的关照机制，不会导致虚伪唤醒，D 错误。
  
• 答案：ACD
  表明：在调用 wait 方法前先锁定互斥量是须要的，由于 wait 方法会在期待时主动开释锁，在被唤醒后重新获取锁，A 精确；调用 notify_one 或 notify_all 方法时不须要持有互斥量，B 错误；使用谓词可以克制虚伪唤醒，确保线程只在条件满足时才继承实验，C 精确；在 wait 方法返回后再次查抄条件可以进一步确保步调的精确性，D 精确。
  

7. 操持标题

7.1 操持一个线程安全的缓存体系

要求：

• 多个线程可查询键是否存在，若不存在则触发盘算任务
  
• 束缚：同一键同时只允许一个盘算任务
  

7.2 实现可取消的并行任务

要求：

• 任务提交后返回future，允许调用方取消正在实验的任务
  
• 须要思量资源整理和线程停止安全
  

7.3 构建生产者-斲丧者管道

要求：

• 多个生产者和斲丧者通过有界缓冲区通讯
  
• 斲丧者须要优先处置处罚特定范例的消息
  

7.4 操持超时安全的数据库毗连池

要求：

• 线程获取毗连时最多期待500ms，超时返回错误
  
• 毗连空闲凌驾2分钟主动采取
  

7.5 实现异步日志体系

要求：

• 全部日志写入哀求异步实验
  
• 当日志队列积存凌驾1000条时切换为同步模式
  
• 支持刷盘操纵期待全部待写日志完成
  

7.1 参考答案

1. #include <iostream>
2. #include <mutex>
3. #include <future>
4. #include <unordered_map>
6. template<typename Key, typename Value>
7. class SafeCache {
8.     std::mutex mtx;
9.     std::unordered_map<Key, Value> cache;
10.     std::unordered_map<Key, std::shared_future<Value>> pending;
12.     Value compute(const Key& key) {
13.         // 模拟耗时计算
14.         std::this_thread::sleep_for(std::chrono::seconds(1));
15.         return key + "_value";
16.     }
18. public:
19.     Value get(const Key& key) {
20.         std::unique_lock lock(mtx);
21.         if (auto it = cache.find(key); it != cache.end()) {
22.             return it->second;
23.         }
25.         if (auto it = pending.find(key); it != pending.end()) {
26.             auto fut = it->second;
27.             lock.unlock();
28.             return fut.get();
29.         }
31.         auto promise = std::make_shared<std::promise<Value>>();
32.         auto fut = promise->get_future().share();
33.         pending.emplace(key, fut);
35.         lock.unlock();
37.         try  {
38.             auto res = compute(key);
39.             lock.lock();
40.             cache.emplace(key, res);
41.             pending.erase(key);
42.             lock.unlock();
43.             promise->set_value(res);
44.         } catch (...) {
45.             promise->set_exception(std::current_exception());
46.         }
48.         return fut.get();
49.     }
50. };
52. // 测试用例
53. void test_cache() {
54.     SafeCache<std::string, std::string> cache;
55.     auto task = [&](std::string key) {
56.         return cache.get(key);
57.     };
59.     auto fut1 = std::async(std::launch::async, task, "A");
60.     auto fut2 = std::async(std::launch::async, task, "A");
61.    
62.     std::cout << fut1.get() << std::endl; // A_value
63.     std::cout << fut2.get() << std::endl; // A_value (仅一次计算)
64. }
66. int main() {
67.     test_cache();
68.     return 0;
69. }
70. /*
71. 分析:
72. 使用std:mutex保护缓存的访问，防止数据竟争。
73. 当键不存在时，为避免多个线程同时触发计算，使用互斥锁和条件变量来保证单个任务执行。可能需要存储计算中的键，防止重复计算。可以使用std::unordered_map存储键和对应的future，表示计算中的结果:
74. 步骤:
75. 1.定义缓存数据结构，包括互斥锁、存储键值对以及正在计算的键及其future。
76. 2.查询缓存时锁住互斥锁，若存在则直接返回值。
77. 3.如果键不存在且未被计算，则生成一个promise和future，存储到正在计算的map中，并启动异步任务计算结果
78. 4.其他线程查询同一键时，等待该键对应的future就绪。
79. 测试用例:
80. 多线程同时请求相同的键，确保只有一个计算任务执行
81. 验证缓存结果是否正确，避免重复计算。
82. 处理计算中的异常情况。
83. */

复制代码

7.2 参考答案

1. #include <iostream>
2. #include <future>
3. #include <atomic>
4. #include <chrono>
5. #include <functional>
7. // 自定义可取消的任务包装类
8. template <typename Result>
9. class CancellableTask {
10. public:
11.     // 构造函数，接受一个可调用对象和一个取消标志
12.     template <typename Func>
13.     CancellableTask(Func func, std::shared_ptr<std::atomic<bool>> cancelFlag)
14.         : cancelFlag_(cancelFlag) {
15.         // 使用 std::async 异步执行任务
16.         future_ = std::async(std::launch::async, [func, cancelFlag]() {
17.             while (!cancelFlag->load()) {
18.                 // 执行任务函数
19.                 Result result = func();
20.                 return result;
21.             }
22.             // 如果任务被取消，抛出异常
23.             throw std::runtime_error("Task cancelled");
24.         });
25.     }
27.     // 获取任务的 future 对象
28.     std::future<Result> get_future() {
29.         return std::move(future_);
30.     }
32.     // 取消任务
33.     void cancel() {
34.         cancelFlag_->store(true);
35.     }
37. private:
38.     std::future<Result> future_;
39.     std::shared_ptr<std::atomic<bool>> cancelFlag_;
40. };
42. // 示例任务函数
43. int exampleTask() {
44.     for (int i = 0; i < 5; ++i) {
45.         std::this_thread::sleep_for(std::chrono::seconds(1));
46.         std::cout << "Task is working: " << i << std::endl;
47.     }
48.     return 42;
49. }
51. int main() {
52.     // 创建一个共享的原子布尔变量作为取消标志
53.     auto cancelFlag = std::make_shared<std::atomic<bool>>(false);
54.     // 创建可取消的任务
55.     CancellableTask<int> task(exampleTask, cancelFlag);
56.     // 获取任务的 future 对象
57.     auto future = task.get_future();
59.     // 模拟一段时间后取消任务
60.     std::this_thread::sleep_for(std::chrono::seconds(2));
61.     std::cout << "Cancelling the task..." << std::endl;
62.     task.cancel();
64.     try {
65.         // 尝试获取任务的结果
66.         int result = future.get();
67.         std::cout << "Task result: " << result << std::endl;
68.     } catch (const std::runtime_error& e) {
69.         std::cout << "Exception caught: " << e.what() << std::endl;
70.     }
72.     return 0;
73. }

复制代码

7.3 参考答案

1. #include <iostream>
2. #include <vector>
3. #include <queue>
4. #include <thread>
5. #include <mutex>
6. #include <condition_variable>
7. #include <algorithm>
8. #include <chrono>
10. // 定义消息类型
11. const int PRIORITY_MESSAGE_TYPE = 1;
13. // 有界缓冲区类
14. template<typename T>
15. class BoundedBuffer {
16. private:
17.     std::vector<T> buffer;
18.     std::mutex mtx;
19.     std::condition_variable not_full;
20.     std::condition_variable not_empty;
21.     size_t capacity;
23. public:
24.     BoundedBuffer(size_t cap) : capacity(cap) {}
26.     // 生产者向缓冲区添加消息
27.     void enqueue(const T& item) {
28.         std::unique_lock<std::mutex> lock(mtx);
29.         // 等待缓冲区有空间
30.         not_full.wait(lock, [this] { return buffer.size() < capacity; });
31.         buffer.push_back(item);
32.         // 通知消费者缓冲区有新消息
33.         not_empty.notify_one();
34.     }
36.     // 消费者从缓冲区取出消息，优先处理特定类型的消息
37.     T dequeue() {
38.         std::unique_lock<std::mutex> lock(mtx);
39.         // 等待缓冲区有消息
40.         not_empty.wait(lock, [this] { return!buffer.empty(); });
42.         // 优先查找特定类型的消息
43.         auto it = std::find_if(buffer.begin(), buffer.end(), [](const T& msg) {
44.             return std::get<1>(msg) == PRIORITY_MESSAGE_TYPE;
45.         });
47.         T item;
48.         if (it != buffer.end()) {
49.             item = *it;
50.             buffer.erase(it);
51.         } else {
52.             item = buffer.front();
53.             buffer.erase(buffer.begin());
54.         }
56.         // 通知生产者缓冲区有空间
57.         not_full.notify_one();
58.         return item;
59.     }
60. };
62. // 生产者函数
63. void producer(BoundedBuffer<std::pair<int, int>>& buffer, int id) {
64.     for (int i = 0; i < 5; ++i) {
65.         int messageType = (i % 2 == 0)? PRIORITY_MESSAGE_TYPE : 2;
66.         buffer.enqueue({id * 10 + i, messageType});
67.         std::cout << "Producer " << id << " produced message " << id * 10 + i << " of type " << messageType << std::endl;
68.         std::this_thread::sleep_for(std::chrono::milliseconds(100));
69.     }
70. }
72. // 消费者函数
73. void consumer(BoundedBuffer<std::pair<int, int>>& buffer, int id) {
74.     for (int i = 0; i < 5; ++i) {
75.         auto [messageId, messageType] = buffer.dequeue();
76.         std::cout << "Consumer " << id << " consumed message " << messageId << " of type " << messageType << std::endl;
77.         std::this_thread::sleep_for(std::chrono::milliseconds(200));
78.     }
79. }
81. int main() {
82.     const int bufferCapacity = 10;
83.     BoundedBuffer<std::pair<int, int>> buffer(bufferCapacity);
85.     const int numProducers = 3;
86.     const int numConsumers = 2;
88.     std::vector<std::thread> producers;
89.     std::vector<std::thread> consumers;
91.     // 创建生产者线程
92.     for (int i = 0; i < numProducers; ++i) {
93.         producers.emplace_back(producer, std::ref(buffer), i);
94.     }
96.     // 创建消费者线程
97.     for (int i = 0; i < numConsumers; ++i) {
98.         consumers.emplace_back(consumer, std::ref(buffer), i);
99.     }
101.     // 等待生产者线程结束
102.     for (auto& p : producers) {
103.         p.join();
104.     }
106.     // 等待消费者线程结束
107.     for (auto& c : consumers) {
108.         c.join();
109.     }
111.     return 0;
112. }

复制代码

7.4 参考答案

1. #include <iostream>
2. #include <queue>
3. #include <thread>
4. #include <mutex>
5. #include <condition_variable>
6. #include <chrono>
7. #include <vector>
9. // 模拟数据库连接类
10. class DatabaseConnection {
11. public:
12.     DatabaseConnection() {
13.         std::cout << "Database connection created." << std::endl;
14.     }
16.     ~DatabaseConnection() {
17.         std::cout << "Database connection destroyed." << std::endl;
18.     }
20.     void executeQuery(const std::string& query) {
21.         std::cout << "Executing query: " << query << std::endl;
22.     }
23. };
25. // 数据库连接池类
26. class DatabaseConnectionPool {
27. private:
28.     std::queue<DatabaseConnection*> connections;
29.     std::mutex mtx;
30.     std::condition_variable cv;
31.     size_t maxConnections;
32.     std::chrono::time_point<std::chrono::steady_clock> lastUsed;
34.     // 自动回收线程函数
35.     void autoReclaim() {
36.         while (true) {
37.             std::this_thread::sleep_for(std::chrono::minutes(1)); // 每分钟检查一次
38.             {
39.                 std::unique_lock<std::mutex> lock(mtx);
40.                 auto now = std::chrono::steady_clock::now();
41.                 while (!connections.empty()) {
42.                     auto connection = connections.front();
43.                     auto idleTime = std::chrono::duration_cast<std::chrono::minutes>(now - lastUsed);
44.                     if (idleTime.count() >= 2) {
45.                         connections.pop();
46.                         delete connection;
47.                     } else {
48.                         break;
49.                     }
50.                 }
51.             }
52.         }
53.     }
55. public:
56.     DatabaseConnectionPool(size_t maxConns) : maxConnections(maxConns) {
57.         for (size_t i = 0; i < maxConns; ++i) {
58.             connections.push(new DatabaseConnection());
59.         }
60.         lastUsed = std::chrono::steady_clock::now();
61.         std::thread(&DatabaseConnectionPool::autoReclaim, this).detach();
62.     }
64.     ~DatabaseConnectionPool() {
65.         std::unique_lock<std::mutex> lock(mtx);
66.         while (!connections.empty()) {
67.             delete connections.front();
68.             connections.pop();
69.         }
70.     }
72.     // 获取数据库连接
73.     DatabaseConnection* getConnection() {
74.         std::unique_lock<std::mutex> lock(mtx);
75.         if (cv.wait_for(lock, std::chrono::milliseconds(500), [this] { return!connections.empty(); })) {
76.             auto connection = connections.front();
77.             connections.pop();
78.             lastUsed = std::chrono::steady_clock::now();
79.             return connection;
80.         }
81.         return nullptr; // 超时返回 nullptr
82.     }
84.     // 释放数据库连接
85.     void releaseConnection(DatabaseConnection* connection) {
86.         std::unique_lock<std::mutex> lock(mtx);
87.         connections.push(connection);
88.         lastUsed = std::chrono::steady_clock::now();
89.         cv.notify_one();
90.     }
91. };
93. // 测试用例
94. void testConnectionPool() {
95.     DatabaseConnectionPool pool(3);
97.     auto task = [&pool]() {
98.         DatabaseConnection* conn = pool.getConnection();
99.         if (conn) {
100.             conn->executeQuery("SELECT * FROM users");
101.             pool.releaseConnection(conn);
102.         } else {
103.             std::cout << "Failed to get a connection (timeout)." << std::endl;
104.         }
105.     };
107.     std::vector<std::thread> threads;
108.     for (int i = 0; i < 5; ++i) {
109.         threads.emplace_back(task);
110.     }
112.     for (auto& t : threads) {
113.         t.join();
114.     }
115. }
117. int main() {
118.     testConnectionPool();
119.     return 0;
120. }

复制代码

7.5 参考答案

1. #include <iostream>
2. #include <fstream>
3. #include <queue>
4. #include <thread>
5. #include <mutex>
6. #include <condition_variable>
7. #include <string>
8. #include <chrono>
9. #include <atomic>
10. #include <vector>
12. // 定义日志级别
13. enum class LogLevel {
14.     DEBUG,
15.     INFO,
16.     WARN,
17.     ERROR
18. };
20. class AsyncLogger {
21. private:
22.     std::queue<std::string> logQueue;
23.     std::mutex mtx;
24.     std::condition_variable cv;
25.     std::ofstream logFile;
26.     std::thread loggerThread;
27.     std::atomic<bool> stopLogging;
28.     std::atomic<bool> isSyncMode;
29.     static constexpr size_t MAX_QUEUE_SIZE = 1000;
31.     // 日志线程函数
32.     void logger() {
33.         std::vector<std::string> buffer;
34.         while (!stopLogging) {
35.             {
36.                 std::unique_lock<std::mutex> lock(mtx);
37.                 cv.wait_for(lock, std::chrono::milliseconds(100), [this] { return!logQueue.empty() || stopLogging; });
38.                 while (!logQueue.empty()) {
39.                     buffer.push_back(std::move(logQueue.front()));
40.                     logQueue.pop();
41.                 }
42.             }
43.             for (const auto& msg : buffer) {
44.                 try {
45.                     logFile << msg << std::endl;
46.                 } catch (const std::exception& e) {
47.                     std::cerr << "Error writing log: " << e.what() << std::endl;
48.                 }
49.             }
50.             buffer.clear();
51.         }
52.         // 处理剩余的日志
53.         {
54.             std::unique_lock<std::mutex> lock(mtx);
55.             while (!logQueue.empty()) {
56.                 try {
57.                     logFile << logQueue.front() << std::endl;
58.                     logQueue.pop();
59.                 } catch (const std::exception& e) {
60.                     std::cerr << "Error writing remaining log: " << e.what() << std::endl;
61.                 }
62.             }
63.         }
64.         try {
65.             logFile.flush();
66.         } catch (const std::exception& e) {
67.             std::cerr << "Error flushing log file: " << e.what() << std::endl;
68.         }
69.     }
71.     // 格式化日志消息，添加日志级别和时间戳
72.     std::string formatLogMessage(LogLevel level, const std::string& message) {
73.         auto now = std::chrono::system_clock::now();
74.         std::time_t currentTime = std::chrono::system_clock::to_time_t(now);
75.         std::string timeStr = std::ctime(&currentTime);
76.         timeStr.pop_back(); // 去掉换行符
77.         std::string levelStr;
78.         switch (level) {
79.             case LogLevel::DEBUG:
80.                 levelStr = "DEBUG";
81.                 break;
82.             case LogLevel::INFO:
83.                 levelStr = "INFO";
84.                 break;
85.             case LogLevel::WARN:
86.                 levelStr = "WARN";
87.                 break;
88.             case LogLevel::ERROR:
89.                 levelStr = "ERROR";
90.                 break;
91.         }
92.         return "[" + timeStr + "] [" + levelStr + "] " + message;
93.     }
95. public:
96.     AsyncLogger(const std::string& logFileName) : stopLogging(false), isSyncMode(false) {
97.         try {
98.             logFile.open(logFileName, std::ios::app);
99.             if (!logFile.is_open()) {
100.                 std::cerr << "Failed to open log file: " << logFileName << std::endl;
101.             }
102.         } catch (const std::exception& e) {
103.             std::cerr << "Error opening log file: " << e.what() << std::endl;
104.         }
105.         loggerThread = std::thread(&AsyncLogger::logger, this);
106.     }
108.     ~AsyncLogger() {
109.         stopLogging = true;
110.         cv.notify_one();
111.         if (loggerThread.joinable()) {
112.             loggerThread.join();
113.         }
114.         try {
115.             logFile.close();
116.         } catch (const std::exception& e) {
117.             std::cerr << "Error closing log file: " << e.what() << std::endl;
118.         }
119.     }
121.     // 写入日志
122.     void log(LogLevel level, const std::string& message) {
123.         std::string formattedMessage = formatLogMessage(level, message);
124.         {
125.             std::unique_lock<std::mutex> lock(mtx);
126.             if (logQueue.size() >= MAX_QUEUE_SIZE) {
127.                 isSyncMode = true;
128.             } else {
129.                 isSyncMode = false;
130.             }
131.         }
133.         if (isSyncMode) {
134.             // 同步模式，直接写入日志
135.             std::lock_guard<std::mutex> lock(mtx);
136.             try {
137.                 logFile << formattedMessage << std::endl;
138.                 logFile.flush();
139.             } catch (const std::exception& e) {
140.                 std::cerr << "Error writing log in sync mode: " << e.what() << std::endl;
141.             }
142.         } else {
143.             // 异步模式，将日志消息放入队列
144.             {
145.                 std::lock_guard<std::mutex> lock(mtx);
146.                 logQueue.push(formattedMessage);
147.             }
148.             cv.notify_one();
149.         }
150.     }
152.     // 刷盘操作
153.     void flush() {
154.         std::unique_lock<std::mutex> lock(mtx);
155.         while (!logQueue.empty()) {
156.             cv.wait(lock, [this] { return logQueue.empty(); });
157.         }
158.         try {
159.             logFile.flush();
160.         } catch (const std::exception& e) {
161.             std::cerr << "Error flushing log during flush operation: " << e.what() << std::endl;
162.         }
163.     }
164. };
166. // 测试用例
167. void testAsyncLogger() {
168.     AsyncLogger logger("test.log");
170.     // 模拟大量日志写入
171.     for (int i = 0; i < 2000; ++i) {
172.         logger.log(LogLevel::INFO, "Log message " + std::to_string(i));
173.         std::cout << "Log message " << i << std::endl;  // 输出到控制台
174.     }
176.     // 刷盘操作
177.     logger.flush();
178. }
180. int main() {
181.     testAsyncLogger();
182.     return 0;
183. }

复制代码