多线程实战：从循环打印ABC到单例模式线程池

时间：2024-10-22 09:32:19

前言

        从进行简单的线程创建到复杂的线程池管理，如何掌握必要的技能来编写现代、高效的并发程序？本文将从大厂面试代码测评常考的循环打印ABC讲起，再到生产者消费者模型的实现，最后结合设计模式和C++11特性的综合应用，手撕单例模式线程池。通过这三个项目的实战，达到大厂对多线程编程的考察标准。

大纲

  1. 多线程基础知识

  2. C++线程库基础

  3. 实战 - 循环打印ABC

  4. 实战 - 生产者消费者模型

  5. 实战 - 单例模式线程池

多线程基础知识

  • 进程和线程的区别

    • 进程是操作系统资源分配的基本单位，是运行中的程序；

    • 线程是任务调度和执行的基本单位，是进程中实际的执行单元，共享进程的地址空间、数据、文件，有自己的寄存器和堆栈，切换开销更小；

      • 地址空间：进程用于内存寻址的一套地址集合，每个进程的地址空间相互独立。通过程序装入内存时的静态重定位或者基址寄存器和界限寄存器实现的动态重定位实现正确访问对应的物理地址。

    • 协程更小，类似于不带返回值的函数调用或者用户级线程。在大量线程场景下可以减少线程频繁调度带来的开销，包括切换上下文消耗的系统时间和存储上下文消耗的内存空间，配合异步IO可以提高性能，适用于IO密集型的任务。（压榨单个线程）缺点是增加编程复杂度。

  • 线程的实现方式

    • 用户级线程

      • 通过线程库实现，由应用程序本身管理和控制线程切换；

      • 用户态下即可完成线程切换，切换开销小，效率高；

      • 只能并发不能并行，一个线程被阻塞进程将被阻塞

    • 内核级线程

      • 由操作系统管理和控制线程切换

      • 需要切换到内核态完成线程切换和上下文切换

      • 可以并行执行

  • 线程常见问题

    • 线程安全问题：多个线程同时访问共享数据时，由于线程间的执行顺序不确定，导致数据的不一致性或状态的错误。需要通过锁机制实现线程间的同步。

    • 死锁问题：多个线程互相等待其它线程释放资源，导致都无法正常执行。

    • 缓存一致性问题：在多核处理器系统中，多个CPU缓存的数据与主内存中的数据未保持同步导致线程读取了错误的数据。

    • 资源泄露：线程请求资源后未正确释放。

C++线程库基础

  • C++线程库模型

        C++线程库在不同的操作系统上可能有不同的实现，Windows上有两种模型：win32和posix，编译组件安装时可选，使用 g++ -v 指令可以查看类型：

  1. Win32线程模型：这是 Windows 特有的线程实现，使用 Windows API 函数，如CreateThread、WaitForSingleObject等来管理线程。如果你选择使用Win32线程模型，std::thread将不可用，但你仍然可以使用Win32 API来创建和管理线程。

  2. POSIX线程模型：POSIX 线程（通常称为 pthreads）是一种跨平台的线程标准，它在多种操作系统上都有实现，包括 Linux 和一些 Unix 系统。在 Windows 上，即使不是原生支持，也可以通过 MinGW-w64 等使用 POSIX 线程。选择 POSIX 线程模型允许你使用std::thread以及C++11标准中的其他线程相关功能。

  • 线程的创建和结束

        在线程销毁前一定要记得设置线程的结束方式，防止意外错误。结束方式有 join 和 detach 两种，其中 join 阻塞方式可以加判断条件 joinable。线程开始执行就成为活动线程，活动线程执行完毕在 join 之前仍然是活动线程，只有活动线程可以 join，活动线程的 joinable 值为 true，join 后变为 false。

        当使用 detach 结束方式时，要注意工作线程是否调用了主线程的变量，因为主线程不会等待其结束就销毁资源，此时工作线程仍调用就会非法访问内存。

#include <threads>

template<typename... Args>
void func(Args... args){
    //线程逻辑
}

std::thread t1(func,args); // 创建线程，参数列表跟在函数参数后面

if(t1.joinable()) t1.join(); // 主线程等待t1执行完再继续执行

t1.detach(); // 分离线程，使其变成一个守护线程独立执行
// 主线程可以继续执行其他任务，而不需要等待t1线程结束

  • 死锁问题

多个线程或进程互相等待对方持有的资源导致都无法正常运行

        如何预防：

破坏互斥条件：允许资源被多个进程共享。

破坏占有和等待条件：要求进程在请求任何资源之前，释放所有已经占有的资源。

破坏不可抢占条件：允许资源被抢占。

破坏循环等待条件：通过资源排序算法，确保每个进程按顺序请求资源。

        如何避免：

银行家算法：记录系统资源和进程状态，包括资源总量、进程的最大需求、已分配资源等，当进程请求资源时，先进行安全性检查，即当前空闲资源能否满足其要求，如果够让它运行才分配。

代码层面：1、需求资源相同的线程，按固定顺序获取锁；

2、需求资源不同的线程，请求资源前先判断当前空闲资源能否满足其执行，若能则一次性请求所有资源，这样能保证始终存在一个线程可以执行，避免陷入死锁。

3、对于已经发生死锁的情况，可以设置资源持有时间，超时释放；也可以在操作系统层面设置死锁检测机制，检测到死锁发生时主动释放部分资源。

  • 如何解决线程安全问题

1、使用互斥量mutex

#include <mutex>

mutex mtx;

void thread1(){
    mtx.lock();
    //线程逻辑
    mtx.unlock();
}

2、使用线程安全函数 call_once

#include <mutex>

once_flag flag;

void func(){
    //线程逻辑
}

void thread1(){
    //保证同时只有一个线程调用函数func
    call_once(flag, func);
}

3、使用封装的锁对象 lock_guard 或者 unique_lock（支持更多功能）

#include <mutex>

mutex mtx;
timed_mutex t_mtx

void thread1(){
    lock_guard<mutex> lock(mtx);
    //线程逻辑
}

void thread2(){
    unique_lock<timed_mutex> lock(t_mtx);
    //最多尝试获取锁2秒，获取不到锁判断为假
    if(lock.try_lock_for(chrono::seconds(2))){
    //线程逻辑
    }
}

  • 条件变量 condition_variable

条件变量在多线程编程中是一个重要的同步机制，主要用于线程之间的通信和协调。

提高CPU利用率：

  • 条件变量允许一个线程在某个条件不满足时进入等待状态，直到其他线程通知它条件已经满足。这种机制避免了忙等待，提高了资源利用率。

实现生产者-消费者模型：

  • 在生产者-消费者问题中，生产者线程可以在缓冲区满时等待，消费者线程在缓冲区空时等待。条件变量使得生产者和消费者能够有效地协调工作。

#include <condition_variable>

condition_variable cv;

void thread1(){
    //获取锁代码
    //条件为假时等待，监听通知
    cv.wait(lock, condition);
    //线程逻辑
}

void thread2(){
    //获取锁代码
    //线程逻辑
    //通知等待线程
    cv.notify_one();
}

实战

  • 三线程循环打印ABC

        三线程共享变量 g_ctr，初值为0，当一个线程获取到锁，发现对3取模不是自己的打印时机，则释放锁并进入等待，直到应该打印的线程获取到锁并打印，将 g_ctr 的值加一，离开当前作用域，自动释放锁，再通知所有其它线程来竞争锁。

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

using namespace std;

mutex mtx;
condition_variable cv;
int flag = 0;

void printChar(char c, int threadId)
{
    for (int i = 0; i < 10; i++)
    {
        unique_lock<mutex> lock(mtx);
        cv.wait(lock, [threadId]
                { return threadId == flag; }); // 等待轮到本线程执行
        cout << c << flush;
        flag = (flag + 1) % 3;
        cv.notify_all(); // 通知其他线程
    }
}

int main()
{
    
    thread t1(printChar, 'A', 0);
    thread t2(printChar, 'B', 1);
    thread t3(printChar, 'C', 2);

    t1.join();
    t2.join();
    t3.join();

    return 0;
}

  • 生产者消费者模型

        任意数量的生产者和消费者协同工作，共享一个任务队列，生产者放消费者取。当任务队列满达到容量上限时生产者等待，并由消费者唤醒；任务队列空时消费者等待，由生产者唤醒。

        实际应用中，生产和消费的过程和操作任务队列的过程最好分离，否则会造成大量的锁的不必要占用。锁外判断条件需在锁内再次考虑，考虑是否需要双重检查以保证唯一性。

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
using namespace std;

mutex g_mtx;//公用锁，任务队列同时只能由一个线程操作
condition_variable g_cv_producer;
condition_variable g_cv_consumer;

queue<int> g_tasks;
int g_taskid = 0;
int tasknum = 100;
int capacity = 10;

void producer() {
    while (g_taskid < tasknum) {
        //实际生产逻辑在这里，生产好了再竞争锁添加任务队列；也可以在锁内生产
        {
            unique_lock<mutex> lock(g_mtx);
            //如果任务队列已满则等待，不满继续
            g_cv_producer.wait(lock, []() {return g_tasks.size() < capacity;});

            //双重检查
            if (g_taskid < tasknum) {
                g_taskid++;
                g_tasks.emplace(g_taskid);
                cout << "producer add task " << g_taskid << endl;
            }
        }
        //生产完唤醒一个消费者进行消费
        g_cv_consumer.notify_one();
    }
}

void consumer() {
    while (g_taskid < tasknum || !g_tasks.empty()) {
        int taskid;
        {
            unique_lock<mutex> lock(g_mtx);
            //如果任务队列为空则等待，非空继续
            g_cv_consumer.wait(lock, []() {return !g_tasks.empty();});
            taskid = g_tasks.front();
            cout << "consumer done task " << taskid << endl;
            g_tasks.pop();
        }
        //取出任务后唤醒一个生产者进行生产
        g_cv_producer.notify_one();
        //实际消费逻辑在这里，取出任务后即释放锁；也可以在锁内消费
    }
}

int main() {
    thread producer_1(producer);
    thread producer_2(producer);
    thread consumer_1(consumer);
    thread consumer_2(consumer);

    producer_1.join();
    producer_2.join();
    consumer_1.join();
    consumer_2.join();

    return 0;
}

  • 单例模式线程池

        单例模式就是将一组全局变量限定作用范围在这个类中，类似命名空间，同时这个类不需要实例化多个对象，只需要提供全局唯一访问点的一种设计模式。在 Java 中一般写单例模式需要使用双检锁和 volatile 关键字，而 C++ 由于从 C++11 开始规定了static变量是线程安全的，保证了 static 变量只被初始化一次，也就不需要写这些内容了。

        具体的线程池的实现，是通过一个任务队列和一组线程来完成的，对外提供添加待执行任务的接口，并使用模版和参数包以及参数绑定的方式来支持任意数量和类型的参数传递。

        在单例模式中，使用 static 声明各个成员，是为了保证对象实例的唯一性，这与显式地禁用拷贝构造函数和赋值构造运算符的目的是一致的。

#include <iostream>
#include <functional>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
using namespace std;

class Threadpool {
private:
    static mutex mtx;
    static condition_variable cv;
    static queue < function<void()>> tasks;//任务队列
    static vector<thread> threads;//一组线程
    static const int thread_nums = 10;

    Threadpool() {
        for (int i = 0;i < thread_nums;++i) {
            threads.emplace_back([this]() {
                function<void()> task;
                while (1) {
                    {
                        unique_lock<mutex> lock(mtx);
                        cv.wait(lock, [this]() {return !tasks.empty();});
                        task = tasks.front();
                        tasks.pop();
                    }
                    task();
                    //实际后台处理逻辑
                }
                });
        }
    }

    ~Threadpool() {
        for (int i = 0;i < thread_nums;++i) {
            threads[i].join();
        }
    }

public:
    Threadpool(const Threadpool&) = delete;
    Threadpool& operator=(const Threadpool&) = delete;

    //添加任务，一个任务为一个特定的函数，使用模板兼容各种函数
    template<typename F, typename... Args>
    static void add_task(F f, Args&&... args) {
        function<void()> task = bind(f, args...);
        {
            unique_lock<mutex> lock(mtx);
            tasks.emplace(task);
        }
        cv.notify_one();
    }

    static Threadpool& getinstance() {
        static Threadpool instance;
        return instance;
    }
};

//静态成员需要类外初始化
mutex Threadpool::mtx;
condition_variable Threadpool::cv;
queue < function<void()>> Threadpool::tasks;
vector<thread> Threadpool::threads;

void log(string msg, int i) {
    cout << msg << i << endl;
}

int main() {
    Threadpool& instance = Threadpool::getinstance();
    for (int i = 0;i < 100; ++i) {
        instance.add_task(log, "task ", i);
    }

    return 0;
}

执行结果：task() 分别在锁内执行和在锁外即后台执行的结果：

The End

参考资料：

C++11实现跨平台线程池

相关文章