Linux C++ 多线程核心原理与工业级实践

Linux C++ 多线程核心原理与工业级实践 | 极客日志

pthread_mutex_lock(&mutex); // 循环检查条件，避免虚假唤醒 while (条件不满足) { pthread_cond_wait(&cond, &mutex); // 自动释放锁，阻塞等待；被唤醒后重新加锁 } // 处理共享资源 pthread_mutex_unlock(&mutex);

#include <iostream>
#include <pthread.h>
#include <string>
#include <stdexcept>
#include <unistd.h>

// 工业级 RAII 封装 pthread 线程
class Thread {
public:
    // 线程函数类型定义
    using ThreadFunc = void* (*)(void*);

    // 构造函数：创建线程，默认设为分离态（避免僵尸线程）
    Thread(ThreadFunc func, void* arg, const std::string& name = "Thread")
        : tid_(0), func_(func), arg_(arg), name_(name), is_running_(false) {
        // 设置线程属性：分离态
        pthread_attr_t attr;
        int ret = pthread_attr_init(&attr);
        if (ret != 0) {
            throw std::runtime_error("pthread_attr_init failed, errno: " + std::to_string(ret));
        }
        ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
        if (ret != 0) {
            pthread_attr_destroy(&attr);
            throw std::runtime_error("pthread_attr_setdetachstate failed, errno: " + std::to_string(ret));
        }
        // 创建线程
        ret = pthread_create(&tid_, &attr, func_, arg_);
        if (ret != 0) {
            pthread_attr_destroy(&attr);
            throw std::runtime_error("pthread_create failed, errno: " + std::to_string(ret));
        }
        pthread_attr_destroy(&attr);
        is_running_ = true;
        std::cout << "Thread " << name_ << " created, tid: " << (unsigned long)tid_ << std::endl;
    }

    // 析构函数：确保线程资源释放（分离态无需 join）
    ~Thread() {
        if (is_running_) {
            std::cout << "Thread " << name_ << " destroyed" << std::endl;
        }
    }

    // 禁止拷贝（线程不可拷贝）
    Thread(const Thread&) = delete;
    Thread& operator=(const Thread&) = delete;

    // 获取线程 ID
    pthread_t tid() const { return tid_; }

private:
    pthread_t tid_;          // 线程 ID
    ThreadFunc func_;        // 线程执行函数
    void* arg_;              // 函数参数
    std::string name_;       // 线程名称（便于调试）
    bool is_running_;        // 线程是否运行
};

// 线程执行函数（打印参数，模拟工作）
void* thread_func(void* arg) {
    std::string* msg = static_cast<std::string*>(arg);
    for (int i = 0; i < 5; ++i) {
        std::cout << "Thread working: " << *msg << ", count: " << i << std::endl;
        sleep(1);
    }
    // 注意：arg 是堆变量，需手动释放
    delete msg;
    pthread_exit(nullptr);
}

int main() {
    try {
        // 传递堆变量，避免栈变量销毁问题
        std::string* msg = new std::string("Hello Linux Thread");
        Thread t(thread_func, msg, "WorkerThread");
        // 主线程等待子线程执行完（分离态线程不影响，仅为演示）
        sleep(6);
    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

g++ -std=c++17 -lpthread thread_raii.cpp -o thread_raii
./thread_raii

#include <iostream>
#include <pthread.h>
#include <queue>
#include <string>
#include <stdexcept>
#include <unistd.h>
#include <atomic>

// 工业级线程安全的队列（生产者消费者核心）
template <typename T>
class SafeQueue {
public:
    SafeQueue(int max_size = 10) : max_size_(max_size), is_running_(true) {
        // 初始化互斥锁和条件变量
        int ret = pthread_mutex_init(&mutex_, nullptr);
        if (ret != 0) {
            throw std::runtime_error("pthread_mutex_init failed: " + std::to_string(ret));
        }
        ret = pthread_cond_init(&not_empty_, nullptr);
        if (ret != 0) {
            pthread_mutex_destroy(&mutex_);
            throw std::runtime_error("pthread_cond_init not_empty failed: " + std::to_string(ret));
        }
        ret = pthread_cond_init(&not_full_, nullptr);
        if (ret != 0) {
            pthread_cond_destroy(&not_empty_);
            pthread_mutex_destroy(&mutex_);
            throw std::runtime_error("pthread_cond_init not_full failed: " + std::to_string(ret));
        }
    }

    ~SafeQueue() {
        // 优雅退出：停止接收新任务，唤醒所有等待线程
        is_running_ = false;
        pthread_cond_broadcast(&not_empty_);
        pthread_cond_broadcast(&not_full_);
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&not_empty_);
        pthread_cond_destroy(&not_full_);
    }

    // 禁止拷贝
    SafeQueue(const SafeQueue&) = delete;
    SafeQueue& operator=(const SafeQueue&) = delete;

    // 生产任务：队列满则阻塞
    bool push(const T& task) {
        pthread_mutex_lock(&mutex_);
        // 循环检查：队列满且运行中则等待
        while (is_running_ && queue_.size() >= max_size_) {
            pthread_cond_wait(&not_full_, &mutex_);
        }
        // 若已停止运行，释放锁并返回
        if (!is_running_) {
            pthread_mutex_unlock(&mutex_);
            return false;
        }
        // 生产任务
        queue_.push(task);
        std::cout << "Producer push task: " << task << ", queue size: " << queue_.size() << std::endl;
        // 唤醒消费者（队列非空）
        pthread_cond_signal(&not_empty_);
        pthread_mutex_unlock(&mutex_);
        return true;
    }

    // 消费任务：队列空则阻塞
    bool pop(T& task) {
        pthread_mutex_lock(&mutex_);
        // 循环检查：队列空且运行中则等待（避免虚假唤醒）
        while (is_running_ && queue_.empty()) {
            pthread_cond_wait(&not_empty_, &mutex_);
        }
        // 若已停止运行且队列为空，释放锁并返回
        if (!is_running_ && queue_.empty()) {
            pthread_mutex_unlock(&mutex_);
            return false;
        }
        // 消费任务
        task = queue_.front();
        queue_.pop();
        std::cout << "Consumer pop task: " << task << ", queue size: " << queue_.size() << std::endl;
        // 唤醒生产者（队列非满）
        pthread_cond_signal(&not_full_);
        pthread_mutex_unlock(&mutex_);
        return true;
    }

    // 停止队列（优雅退出）
    void stop() {
        pthread_mutex_lock(&mutex_);
        is_running_ = false;
        pthread_mutex_unlock(&mutex_);
        pthread_cond_broadcast(&not_empty_);
        pthread_cond_broadcast(&not_full_);
    }

private:
    std::queue<T> queue_;           // 任务队列
    int max_size_;                  // 队列最大容量（避免内存溢出）
    std::atomic<bool> is_running_;  // 队列是否运行（原子变量，避免数据竞争）
    pthread_mutex_t mutex_;         // 互斥锁
    pthread_cond_t not_empty_;      // 队列非空条件变量（唤醒消费者）
    pthread_cond_t not_full_;       // 队列非满条件变量（唤醒生产者）
};

// 生产者线程函数
void* producer_func(void* arg) {
    SafeQueue<std::string>* queue = static_cast<SafeQueue<std::string>*>(arg);
    for (int i = 0; i < 8; ++i) {
        std::string task = "Task-" + std::to_string(i);
        queue->push(task);
        sleep(1); // 模拟生产耗时
    }
    return nullptr;
}

// 消费者线程函数
void* consumer_func(void* arg) {
    SafeQueue<std::string>* queue = static_cast<SafeQueue<std::string>*>(arg);
    std::string task;
    while (queue->pop(task)) {
        // 模拟消费耗时
        sleep(2);
    }
    return nullptr;
}

int main() {
    try {
        SafeQueue<std::string> queue(5); // 队列最大容量 5
        // 创建生产者和消费者线程
        pthread_t producer_tid, consumer_tid;
        pthread_create(&producer_tid, nullptr, producer_func, &queue);
        pthread_create(&consumer_tid, nullptr, consumer_func, &queue);
        // 主线程运行 15 秒后停止队列
        sleep(15);
        queue.stop();
        std::cout << "Main thread stop queue" << std::endl;
        // 等待线程退出
        pthread_join(producer_tid, nullptr);
        pthread_join(consumer_tid, nullptr);
    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

#include <iostream>
#include <pthread.h>
#include <queue>
#include <string>
#include <stdexcept>
#include <unistd.h>
#include <atomic>
#include <functional>
#include <vector>

// 任务类型：可调用对象（支持 lambda、函数指针等）
using Task = std::function<void()>;

// 任务拒绝策略枚举
enum class RejectPolicy {
    BLOCK,    // 阻塞等待
    DISCARD,  // 丢弃任务
    CALLER_RUN // 调用者自行执行
};

// 工业级线程池
class ThreadPool {
public:
    ThreadPool(int core_size = 4, int max_size = 8, int queue_size = 100, RejectPolicy policy = RejectPolicy::BLOCK)
        : core_size_(core_size), max_size_(max_size), queue_size_(queue_size), reject_policy_(policy),
          is_running_(true), current_threads_(0), idle_threads_(0) {
        // 初始化同步资源
        int ret = pthread_mutex_init(&mutex_, nullptr);
        if (ret != 0) throw std::runtime_error("mutex init failed: " + std::to_string(ret));
        ret = pthread_cond_init(&not_empty_, nullptr);
        if (ret != 0) {
            pthread_mutex_destroy(&mutex_);
            throw std::runtime_error("cond not_empty init failed: " + std::to_string(ret));
        }
        ret = pthread_cond_init(&not_full_, nullptr);
        if (ret != 0) {
            pthread_cond_destroy(&not_empty_);
            pthread_mutex_destroy(&mutex_);
            throw std::runtime_error("cond not_full init failed: " + std::to_string(ret));
        }
        // 创建核心线程
        for (int i = 0; i < core_size_; ++i) {
            create_thread();
        }
    }

    ~ThreadPool() {
        // 优雅退出
        is_running_ = false;
        pthread_cond_broadcast(&not_empty_); // 唤醒所有等待任务的线程
        pthread_cond_broadcast(&not_full_);  // 唤醒所有等待队列的线程
        // 等待所有线程退出
        pthread_mutex_lock(&mutex_);
        for (pthread_t tid : threads_) {
            pthread_join(tid, nullptr);
        }
        pthread_mutex_unlock(&mutex_);
        // 释放同步资源
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&not_empty_);
        pthread_cond_destroy(&not_full_);
    }

    // 禁止拷贝
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;

    // 提交任务
    bool submit(const Task& task) {
        pthread_mutex_lock(&mutex_);
        // 检查队列是否满
        while (is_running_ && task_queue_.size() >= queue_size_) {
            switch (reject_policy_) {
                case RejectPolicy::BLOCK:
                    // 阻塞等待队列非满
                    pthread_cond_wait(&not_full_, &mutex_);
                    break;
                case RejectPolicy::DISCARD:
                    // 丢弃任务
                    pthread_mutex_unlock(&mutex_);
                    std::cerr << "Task discarded: queue is full" << std::endl;
                    return false;
                case RejectPolicy::CALLER_RUN:
                    // 调用者自行执行任务
                    pthread_mutex_unlock(&mutex_);
                    task();
                    std::cout << "Task run by caller: queue is full" << std::endl;
                    return true;
            }
        }
        if (!is_running_) {
            pthread_mutex_unlock(&mutex_);
            std::cerr << "ThreadPool is stopped, reject task" << std::endl;
            return false;
        }
        // 添加任务到队列
        task_queue_.push(task);
        std::cout << "Submit task, queue size: " << task_queue_.size() << std::endl;
        // 唤醒空闲线程
        pthread_cond_signal(&not_empty_);
        // 动态扩容：核心线程已满，队列未满，且当前线程数<最大线程数
        if (idle_threads_ == 0 && current_threads_ < max_size_) {
            create_thread();
        }
        pthread_mutex_unlock(&mutex_);
        return true;
    }

private:
    // 创建线程
    void create_thread() {
        pthread_t tid;
        int ret = pthread_create(&tid, nullptr, thread_func, this);
        if (ret != 0) {
            pthread_mutex_unlock(&mutex_);
            throw std::runtime_error("create thread failed: " + std::to_string(ret));
        }
        threads_.push_back(tid);
        current_threads_++;
        idle_threads_++;
        std::cout << "Create thread, current threads: " << current_threads_ << std::endl;
    }

    // 线程执行函数
    static void* thread_func(void* arg) {
        ThreadPool* pool = static_cast<ThreadPool*>(arg);
        pool->run();
        return nullptr;
    }

    // 线程运行逻辑
    void run() {
        while (is_running_) {
            Task task;
            pthread_mutex_lock(&mutex_);
            // 空闲线程数 +1
            idle_threads_++;
            // 等待任务（队列非空或线程池停止）
            while (is_running_ && task_queue_.empty()) {
                std::cout << "Thread " << (unsigned long)pthread_self() << " wait task" << std::endl;
                pthread_cond_wait(&not_empty_, &mutex_);
            }
            // 空闲线程数 -1
            idle_threads_--;
            // 取出任务
            if (!task_queue_.empty()) {
                task = task_queue_.front();
                task_queue_.pop();
                std::cout << "Thread " << (unsigned long)pthread_self() << " take task, queue size: " << task_queue_.size() << std::endl;
                // 唤醒提交任务的线程（队列非满）
                pthread_cond_signal(&not_full_);
            }
            pthread_mutex_unlock(&mutex_);
            // 执行任务（解锁后执行，避免锁持有时间过长）
            if (task) {
                try {
                    task();
                } catch (const std::exception& e) {
                    std::cerr << "Task execute failed: " << e.what() << std::endl;
                }
            }
            // 动态缩容：当前线程数>核心线程数，且空闲时间过长（简化版：直接退出）
            if (current_threads_ > core_size_ && idle_threads_ > core_size_) {
                pthread_mutex_lock(&mutex_);
                current_threads_--;
                // 从线程列表中移除当前线程
                for (auto it = threads_.begin(); it != threads_.end(); ++it) {
                    if (*it == pthread_self()) {
                        threads_.erase(it);
                        break;
                    }
                }
                pthread_mutex_unlock(&mutex_);
                std::cout << "Thread " << (unsigned long)pthread_self() << " exit, current threads: " << current_threads_ << std::endl;
                return;
            }
        }
    }

private:
    int core_size_;             // 核心线程数
    int max_size_;              // 最大线程数
    int queue_size_;            // 任务队列最大容量
    RejectPolicy reject_policy_; // 任务拒绝策略
    std::vector<pthread_t> threads_; // 线程列表
    std::queue<Task> task_queue_;    // 任务队列
    std::atomic<bool> is_running_;   // 线程池是否运行
    int current_threads_;            // 当前线程数
    int idle_threads_;               // 空闲线程数
    pthread_mutex_t mutex_;          // 互斥锁
    pthread_cond_t not_empty_;       // 队列非空条件变量
    pthread_cond_t not_full_;        // 队列非满条件变量
};

// 测试任务：打印任务 ID 和线程 ID
void test_task(int id) {
    std::cout << "Task " << id << " execute by thread: " << (unsigned long)pthread_self() << std::endl;
    sleep(1); // 模拟任务耗时
}

int main() {
    try {
        // 创建线程池：核心 4 线程，最大 8 线程，队列容量 10，阻塞策略
        ThreadPool pool(4, 8, 10, RejectPolicy::BLOCK);
        // 提交 20 个任务
        for (int i = 0; i < 20; ++i) {
            pool.submit(std::bind(test_task, i));
        }
        // 主线程运行 10 秒后退出
        sleep(10);
    } catch (const std::exception& e) {
        std::cerr << "ThreadPool error: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

# 启动程序
gdb ./thread_pool
# 运行程序
run
# 查看所有线程
info threads
# 切换到指定线程（如线程 2）
thread 2
# 查看线程调用栈
bt
# 查看线程持有的锁
info mutex
# 设置断点
break ThreadPool::submit
# 继续运行
c

# 用 pstack 查看线程栈
pstack <pid>
# 用 valgrind 检测死锁和数据竞争
valgrind --tool=helgrind ./thread_pool
# 用 valgrind 检测内存泄漏
valgrind --tool=memcheck --leak-check=full ./thread_pool

# 用 perf 分析 CPU 占用
perf top -p <pid>
# 用 htop 查看线程 CPU 占用
htop -p <pid>

维度	进程	线程
资源占用	大（独立地址空间、页表等）	小（共享进程资源，仅私有栈/寄存器）
创建/销毁开销	高（需分配地址空间、初始化资源）	低（仅需分配栈空间，复用进程资源）
通信方式	复杂（管道、消息队列、Socket 等）	简单（共享内存、全局变量，需同步）
调度开销	高（上下文切换需切换地址空间）	低（仅切换寄存器/栈，地址空间不变）
独立性	高（一个进程崩溃不影响其他）	低（一个线程崩溃导致整个进程崩溃）
适用场景	独立任务（如多个应用程序）	高并发、数据共享的任务（如服务器处理请求）

维度	互斥锁	自旋锁
等待方式	线程阻塞（放弃 CPU）	线程忙等（循环检测锁是否释放）
开销	上下文切换开销（高）	CPU 空转开销（低）
适用场景	锁持有时间长（如 IO 操作、复杂计算）	锁持有时间极短（如几行内存操作）
资源占用	阻塞时释放 CPU，资源占用低	忙等时占用 CPU，资源占用高

Linux C++ 多线程核心原理与工业级实践

一、多线程基础原理

1. 线程是什么？

2. 进程 vs 线程

3. Linux C++ 多线程核心概念

二、补充知识点

1. Linux 下创建线程的核心 API 有哪些？作用分别是什么？

2. 线程的分离态（detached）和结合态（joinable）有什么区别？

3. 什么是线程安全？如何保证线程安全？

4. 死锁的 4 个必要条件？如何避免/排查死锁？

5. 互斥锁（mutex）vs 自旋锁（spinlock）：选型依据？

6. 条件变量（condition variable）的正确使用方式？为什么必须搭配互斥锁？

7. 线程池的核心原理与工业级设计要点？

三、从基础 API 到生产级代码

1. 环境与规范

2. 实战 1：安全的线程创建与退出（RAII 封装）

3. 实战 2：工业级生产者消费者模型（互斥锁 + 条件变量）

4. 实战 3：工业级线程池实现（支持动态扩容、优雅退出）

5. 扩展方向

四、工业级多线程问题排查：工具与实战

1. GDB 调试多线程

2. 死锁检测

3. 性能分析

更多推荐文章

相关免费在线工具

Linux C++ 多线程核心原理与工业级实践

一、多线程基础原理

1. 线程是什么？

2. 进程 vs 线程

3. Linux C++ 多线程核心概念

二、补充知识点

1. Linux 下创建线程的核心 API 有哪些？作用分别是什么？

2. 线程的分离态（detached）和结合态（joinable）有什么区别？

3. 什么是线程安全？如何保证线程安全？

4. 死锁的 4 个必要条件？如何避免/排查死锁？

5. 互斥锁（mutex）vs 自旋锁（spinlock）：选型依据？

6. 条件变量（condition variable）的正确使用方式？为什么必须搭配互斥锁？

7. 线程池的核心原理与工业级设计要点？

三、从基础 API 到生产级代码

1. 环境与规范

2. 实战 1：安全的线程创建与退出（RAII 封装）

3. 实战 2：工业级生产者消费者模型（互斥锁 + 条件变量）

4. 实战 3：工业级线程池实现（支持动态扩容、优雅退出）

5. 扩展方向

四、工业级多线程问题排查：工具与实战

1. GDB 调试多线程

2. 死锁检测

3. 性能分析

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具