C++ 多线程调试技巧：死锁、活锁、饥饿的定位与分析

好的，各位观众，欢迎来到“C++多线程调试奇遇记”现场！今天咱们聊聊C++多线程里那些磨人的小妖精：死锁、活锁、饥饿。别怕，我会用最接地气的方式，带你们把它们揪出来，暴打一顿！

第一章：死锁——你锁我，我锁你，大家一起原地去世

死锁，顾名思义，就是大家互相锁住对方需要的资源，谁也动不了。就像一群人围成圈，每个人都抓住前面人的衣服，谁也走不了。

1. 死锁的成因：

要理解死锁，先要了解它产生的几个必要条件，这就像是犯罪的动机和作案工具：

• 互斥条件： 资源是独占的，一个资源一次只能被一个线程持有。这就像厕所，只能一个人用，别人得等着。
• 占有且等待条件： 线程占有了一些资源，还在等待其他线程释放它需要的资源。就像你占着茅坑，还等着别人给你递纸。
• 不可剥夺条件： 线程已经获得的资源，在未使用完之前，不能被其他线程强行剥夺。 就像你占着茅坑，别人不能把你拽出来。
• 循环等待条件： 形成一个循环链，每个线程都在等待下一个线程释放资源。 这就像几个人同时上厕所，每个人都堵在另一个人的门口。

只有这四个条件都满足了，才有可能发生死锁。记住，是“有可能”，不是一定。

2. 死锁的例子：

来个最经典的死锁例子，两个线程，两个互斥锁：

#include <iostream>
#include <thread>
#include <mutex>

std::mutex mutex1;
std::mutex mutex2;

void thread1_func() {
    std::cout << "Thread 1: Trying to acquire mutex1..." << std::endl;
    mutex1.lock();
    std::cout << "Thread 1: Acquired mutex1" << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟一些操作
    std::cout << "Thread 1: Trying to acquire mutex2..." << std::endl;
    mutex2.lock();
    std::cout << "Thread 1: Acquired mutex2" << std::endl;

    // ... 使用 mutex1 和 mutex2 保护的资源 ...

    mutex2.unlock();
    mutex1.unlock();
    std::cout << "Thread 1: Released mutex1 and mutex2" << std::endl;
}

void thread2_func() {
    std::cout << "Thread 2: Trying to acquire mutex2..." << std::endl;
    mutex2.lock();
    std::cout << "Thread 2: Acquired mutex2" << std::endl;
    std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 模拟一些操作
    std::cout << "Thread 2: Trying to acquire mutex1..." << std::endl;
    mutex1.lock();
    std::cout << "Thread 2: Acquired mutex1" << std::endl;

    // ... 使用 mutex1 和 mutex2 保护的资源 ...

    mutex1.unlock();
    mutex2.unlock();
    std::cout << "Thread 2: Released mutex1 and mutex2" << std::endl;
}

int main() {
    std::thread thread1(thread1_func);
    std::thread thread2(thread2_func);

    thread1.join();
    thread2.join();

    std::cout << "Program finished" << std::endl;
    return 0;
}

运行这段代码，很大概率你会发现程序卡住了，不再输出任何东西。这就是死锁！线程1拿到了 mutex1，想拿 mutex2；线程2拿到了 mutex2，想拿 mutex1。大家互相等着，谁也不让谁，最终一起去世。

3. 如何定位死锁：

• GDB调试： 在GDB里，你可以查看每个线程的堆栈信息，看看它们都在等待哪个锁。

  • info threads：查看所有线程的信息。
  • thread <线程ID>：切换到指定的线程。
  • bt：查看当前线程的堆栈信息，可以找到线程卡在哪个锁上。

• 线程监控工具： 有一些工具可以监控线程的状态，例如Visual Studio的并发可视化工具，或者Linux下的perf工具。

• 日志： 在代码中加入日志，记录每个线程获取和释放锁的时间和顺序。这就像给犯罪现场拍照，方便事后分析。

4. 如何避免死锁：

• 避免循环等待： 这是最常见的解决死锁的方法。给所有的锁分配一个唯一的编号，然后让所有线程都按照编号顺序获取锁。 就像排队买票，谁也不插队。

  // 假设 mutex1 的编号小于 mutex2
  void thread1_func() {
      std::lock(mutex1, mutex2); // 一次性获取两个锁，避免死锁
      std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
      std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
  
      // ... 使用 mutex1 和 mutex2 保护的资源 ...
  }
  
  void thread2_func() {
      std::lock(mutex1, mutex2); // 同样按照编号顺序获取锁
      std::lock_guard<std::mutex> lock1(mutex1, std::adopt_lock);
      std::lock_guard<std::mutex> lock2(mutex2, std::adopt_lock);
  
      // ... 使用 mutex1 和 mutex2 保护的资源 ...
  }

• 使用 std::unique_lock 和 std::try_lock： std::unique_lock 提供了更灵活的锁管理，可以手动控制锁的获取和释放。 std::try_lock 尝试获取锁，如果获取不到立即返回，不会阻塞。

  void thread1_func() {
      std::unique_lock<std::mutex> lock1(mutex1, std::defer_lock);
      std::unique_lock<std::mutex> lock2(mutex2, std::defer_lock);
  
      while (true) {
          if (std::try_lock(lock1, lock2) == -1) { // 尝试同时获取两个锁
              // 获取成功
              break;
          } else {
              // 获取失败，释放已获取的锁，稍后再试
              lock1.unlock();
              lock2.unlock();
              std::this_thread::sleep_for(std::chrono::milliseconds(10));
          }
      }
  
      // ... 使用 mutex1 和 mutex2 保护的资源 ...
  
      lock1.unlock();
      lock2.unlock();
  }

• 超时机制： 给锁的获取设置一个超时时间，如果超过时间还没获取到锁，就放弃，释放已经获取的锁，避免一直等待。

  void thread1_func() {
      std::unique_lock<std::mutex> lock1(mutex1);
      if (lock1.owns_lock()) {
          std::unique_lock<std::mutex> lock2(mutex2, std::defer_lock);
          if (lock2.try_lock_for(std::chrono::milliseconds(100))) { // 尝试在100ms内获取锁
              // 获取成功
              // ... 使用 mutex1 和 mutex2 保护的资源 ...
          } else {
              // 获取失败，放弃
              std::cout << "Thread 1: Failed to acquire mutex2 within timeout" << std::endl;
          }
      } else {
          std::cout << "Thread 1: Failed to acquire mutex1" << std::endl;
      }
  }

• 减少锁的粒度： 尽量让每个锁保护的代码块更小，减少线程持有锁的时间，降低死锁的概率。

• 使用无锁数据结构： 如果可以，尽量使用无锁数据结构，例如原子变量、无锁队列等，避免使用锁。

第二章：活锁——你让，我让，大家一起跳华尔兹

活锁，跟死锁有点像，但又不太一样。死锁是大家彻底不动了，活锁是大家一直在动，但谁也完不成任务。就像一群人同时要通过一个狭窄的通道，每个人都想让对方先走，结果大家谁也走不出去，一直在互相礼让。

1. 活锁的成因：

活锁通常是因为线程在检测到冲突后，会主动释放资源，然后重试。但是，如果重试的策略不合理，可能会导致大家一直在互相让步，谁也无法前进。

2. 活锁的例子：

#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>

std::mutex mutex;
std::atomic<bool> continue_trying(true);

void thread_func(int thread_id) {
    int attempts = 0;
    while (continue_trying) {
        std::unique_lock<std::mutex> lock(mutex, std::try_to_lock);
        if (lock.owns_lock()) {
            std::cout << "Thread " << thread_id << ": Acquired the lock after " << attempts << " attempts." << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟一些操作
            std::cout << "Thread " << thread_id << ": Released the lock." << std::endl;
            break; // 获取到锁，完成任务，退出循环
        } else {
            std::cout << "Thread " << thread_id << ": Failed to acquire the lock, retrying..." << std::endl;
            attempts++;
            std::this_thread::sleep_for(std::chrono::milliseconds(10)); // 稍微等待一下再重试
        }
    }

    if (!continue_trying) {
        std::cout << "Thread " << thread_id << ": Exiting due to external signal." << std::endl;
    }
}

int main() {
    std::thread thread1(thread_func, 1);
    std::thread thread2(thread_func, 2);

    std::this_thread::sleep_for(std::chrono::seconds(5)); // 运行一段时间
    std::cout << "Main thread: Signaling threads to stop trying." << std::endl;
    continue_trying = false; // 停止重试
    thread1.join();
    thread2.join();

    std::cout << "Program finished." << std::endl;
    return 0;
}

在这个例子中，两个线程都在尝试获取同一个锁，如果获取失败，就立即释放，然后重试。如果两个线程同时尝试获取锁，并且都失败了，它们会同时释放锁，然后同时重试，导致谁也无法成功获取锁。

3. 如何定位活锁：

• 观察线程状态： 观察线程的状态，看看它们是否一直在运行，但却没有任何进展。
• 日志： 在代码中加入日志，记录线程尝试获取锁的次数和时间。如果发现线程一直在重试，但却始终无法获取锁，就可能是活锁。
• 性能监控工具： 使用性能监控工具，例如perf，查看CPU的使用率。如果CPU使用率很高，但程序的实际吞吐量却很低，就可能是活锁。

4. 如何避免活锁：

• 引入随机性： 在重试之前，引入随机的等待时间，避免线程同时重试。 这就像让大家猜拳，谁赢了谁先走。

  #include <random>
  
  void thread_func(int thread_id) {
      std::random_device rd;
      std::mt19937 gen(rd());
      std::uniform_int_distribution<> distrib(10, 50); // 随机等待 10-50 毫秒
  
      int attempts = 0;
      while (continue_trying) {
          std::unique_lock<std::mutex> lock(mutex, std::try_to_lock);
          if (lock.owns_lock()) {
              std::cout << "Thread " << thread_id << ": Acquired the lock after " << attempts << " attempts." << std::endl;
              std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 模拟一些操作
              std::cout << "Thread " << thread_id << ": Released the lock." << std::endl;
              break; // 获取到锁，完成任务，退出循环
          } else {
              std::cout << "Thread " << thread_id << ": Failed to acquire the lock, retrying..." << std::endl;
              attempts++;
              std::this_thread::sleep_for(std::chrono::milliseconds(distrib(gen))); // 随机等待
          }
      }
  
      if (!continue_trying) {
          std::cout << "Thread " << thread_id << ": Exiting due to external signal." << std::endl;
      }
  }

• 限制重试次数： 限制线程重试的次数，如果超过次数仍然无法获取锁，就放弃，或者采取其他措施。

• 优先级反转： 如果高优先级线程一直在等待低优先级线程释放锁，可以考虑提升低优先级线程的优先级，让它尽快完成任务，释放锁。

第三章：饥饿——我等啊等啊等啊，花儿都谢了

饥饿，指的是某个线程因为优先级太低，或者总是被其他线程抢占资源，导致长时间无法获得运行机会。就像一群人排队吃饭，总有人被挤到最后面，永远也吃不上饭。

1. 饥饿的成因：

• 优先级调度： 如果高优先级线程总是抢占低优先级线程的资源，低优先级线程可能会一直处于饥饿状态。
• 不公平的调度算法： 某些调度算法可能导致某些线程总是无法获得运行机会。
• 资源竞争： 如果某个线程需要的资源总是被其他线程占用，它可能会一直处于饥饿状态。

2. 饥饿的例子：

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

std::mutex mutex;
std::condition_variable cv;
bool data_ready = false;

void worker_thread() {
    std::unique_lock<std::mutex> lock(mutex);
    cv.wait(lock, []{ return data_ready; }); // 等待数据准备好
    std::cout << "Worker thread: Processing data..." << std::endl;
    // ... 处理数据的代码 ...
}

void producer_thread() {
    std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟一些耗时操作
    {
        std::lock_guard<std::mutex> lock(mutex);
        data_ready = true;
    }
    std::cout << "Producer thread: Data is ready!" << std::endl;
    cv.notify_one(); // 通知一个等待的线程
}

int main() {
    std::thread worker1(worker_thread);
    std::thread worker2(worker_thread);
    std::thread producer(producer_thread);

    worker1.join();
    worker2.join();
    producer.join();

    std::cout << "Program finished." << std::endl;
    return 0;
}

在这个例子中，如果 worker1 线程先被调度，它会进入 cv.wait() 等待数据准备好。然后 producer 线程准备好数据，并调用 cv.notify_one() 唤醒一个等待的线程。但是，如果 worker2 线程的优先级更高，它可能会抢占 worker1 线程的运行机会，导致 worker2 线程被唤醒，而 worker1 线程仍然在等待，处于饥饿状态。 虽然最终 worker1 也能被唤醒，但是如果系统中有持续的高优先级线程，那么低优先级的线程可能很久都无法得到执行。

3. 如何定位饥饿：

• 观察线程状态： 观察线程的状态，看看是否有线程长时间处于等待状态，或者运行时间明显少于其他线程。
• 性能监控工具： 使用性能监控工具，例如perf，查看每个线程的CPU使用时间。如果发现某个线程的CPU使用时间明显少于其他线程，就可能是饥饿。
• 日志： 在代码中加入日志，记录每个线程的开始时间和结束时间。如果发现某个线程的运行时间很长，但实际完成的工作却很少，就可能是饥饿。

4. 如何避免饥饿：

• 避免优先级反转： 尽量避免高优先级线程等待低优先级线程释放资源。如果无法避免，可以考虑提升低优先级线程的优先级。
• 使用公平的调度算法： 选择公平的调度算法，例如轮询调度，保证每个线程都有机会获得运行机会。
• 限制线程的运行时间： 限制线程的运行时间，避免某个线程长时间占用资源，导致其他线程饥饿。
• 使用 std::condition_variable::notify_all： 如果使用条件变量，尽量使用 notify_all 唤醒所有等待的线程，而不是 notify_one，保证每个线程都有机会被唤醒。

总结：

问题	成因	定位方法	避免方法
死锁	循环等待，互斥，占有且等待，不可剥夺	GDB调试，线程监控工具，日志	避免循环等待，使用 std::try_lock，超时机制，减少锁的粒度，使用无锁数据结构
活锁	线程不断重试，但始终无法获取资源	观察线程状态，日志，性能监控工具	引入随机性，限制重试次数，优先级反转
饥饿	优先级调度，不公平的调度算法，资源竞争	观察线程状态，性能监控工具，日志	避免优先级反转，使用公平的调度算法，限制线程的运行时间，使用 std::condition_variable::notify_all

多线程编程就像走钢丝，一不小心就会掉进坑里。但是，只要我们掌握了这些技巧，就可以轻松应对死锁、活锁和饥饿这些小妖精，让我们的C++程序跑得更快、更稳！ 记住，调试多线程程序需要耐心和细心，多用工具，多打日志，总能找到问题的根源。

好了，今天的讲座就到这里，感谢大家的收看，我们下期再见！