C++ 基于 pthread 的线程池设计与实现

线程管理

我们使用 _q 队列来表示任务队列，既然存在很多线程从任务队列里取数据，那这不就是多线程并发访问共享资源吗？因此，需要对共享资源做保护啊！如何保护呢？互斥锁。可是，该加一把锁呢还是，这里以一把锁就可以解决多线程并发访问共享资源的问题。

#pragma once
#include <pthread.h>
#include <iostream>
#include <unistd.h>
#include <string>

class Mutex {
public:
    Mutex() {
        // 初始化锁
        pthread_mutex_init(&_lock, nullptr);
    }
    void Lock() {
        // 加锁
        pthread_mutex_lock(&_lock);
    }
    void Unlock() {
        // 解锁
        pthread_mutex_unlock(&_lock);
    }
    pthread_mutex_t *Get() { return &_lock; }
    ~Mutex() {
        // 毁坏锁
        pthread_mutex_destroy(&_lock);
    }
private:
    pthread_mutex_t _lock;
};

class LockGuard {
public:
    LockGuard(Mutex *mutex) : _mutexp(mutex) {
        _mutexp->Lock();
    }
    ~LockGuard() {
        _mutexp->Unlock();
    }
private:
    Mutex *_mutexp;
};

可是，单单只有锁是不够的，会存在这种问题。如果某个线程的竞争能力特别强呢？我用户都没往任务队列里放数据，你这个线程就不断加锁，检查任务队列是否有数据，在解锁，这反而会导致效率降低。那么能不能当用户往任务队列里放数据后，再通知线程们去执行这个任务呢？当然可以，如果没有放数据，此时线程们都需要进入休眠，需要使用到条件变量。

#pragma once
#include<pthread.h>
#include"Mutex.hpp"

class Cond {
public:
    Cond() {
        pthread_cond_init(&_cond,nullptr);
    }
    void Wait(Mutex &mutex) {
        int n = pthread_cond_wait(&_cond,mutex.Get());
    }
    void NotifyOne() {
        int n = pthread_cond_signal(&_cond);
        (void)n;
    }
    void NotifyAll() {
        int n = pthread_cond_broadcast(&_cond);
        (void)n;
    }
    ~Cond() {
        pthread_cond_destroy(&_cond);
    }
private:
    pthread_cond_t _cond;
};

线程池实现

前置工作准备完毕，那么就可以开始对线程池的封装了！

template <class T> class ThreadPool {
private:
    // 任务队列
    std::queue<T> _q;
    // 整体使用的临界资源
    // 多个线程
    std::vector<Thread> _threads;
    int _threadnum;
    int _wait_thread_num;
    // 保护机制
    Mutex _lock;
    Cond _cond;
    // 其他属性
    bool _is_running;
};

这里的 std::vector<Thread> _threads 是否创建了线程对象呢？可以肯定的是肯定是没有的，相当于你只是拥有了线程这个图纸，里面含有线程的相关属性。

构造函数

// 线程池
const static int defaultthreadnum = 3; // for debug

template <class T> class ThreadPool {
private:
    void Routine(const std::string &name) {
        while (true) {
            // 把任务从线程获取到线程私有！临界区 -> 私有的栈
            T t;
            {
                LockGuard lockguard(&_lock);
                while (QueueIsEmpty() && _is_running) {
                    _wait_thread_num++;
                    _cond.Wait(_lock);
                    _wait_thread_num--;
                }
                if (!_is_running && QueueIsEmpty()) {
                    LOG(LogLevel::INFO) << " 线程池退出 && 任务队列为空，" << name << " 退出";
                    break;
                }
                // 队列中一定有任务了!, 但是
                // 1. 线程池退出 -- 消耗历史
                // 2. 线程池没有退出 -- 正常工作
                t = _q.front();
                _q.pop();
            }
            t(); // 规定，未来的任务，必须这样处理！
            // for debug
            // LOG(LogLevel::DEBUG) << name << " handler task: " << t.Result2String();
        }
    }

public:
    ThreadPool(int threadnum = defaultthreadnum) 
        : _threadnum(threadnum), _is_running(false), _wait_thread_num(0) {
        for (int i = 0; i < _threadnum; i++) {
            std::string name = "thread-" + std::to_string(i + 1);
            _threads.emplace_back([this](const std::string &name) { 
                this->Routine(name); 
            }, name);
        }
        LOG(LogLevel::INFO) << "thread pool obj create success";
    }
};

问题 1: 构造函数这里的 for 循环时对象存在了吗？ 存在了的，在初始化列表这里对对象进行初始化，进行了开辟空间，已然可以访问类内属性。

问题 2: 构造函数并不需要启动线程池，只负责获取资源并设置其要执行的任务。只有当 ThreadPool 对象调用 Start 成员函数时，才需要让线程池启动起来。

我们的想法是这样的：在这里将 this 指针与任务执行方法进行绑定，当线程池启动起来时，能够调用 Thread 线程对象的 start 函数，里面的 pthread_create 函数会调用 start_routine 函数，函数体内的 self->_func(self->_name) 会因为包装器的原因，回调到 ThreadPool 的任务执行方法。这种思想可以做到模块与模块之间的解耦合。

线程启动、取消与回收

线程启动

void Start() {
    if (_is_running) return;
    _is_running = true;
    for (auto &t : _threads) {
        t.Start();
    }
    LOG(LogLevel::INFO) << "thread pool running success";
}

线程取消

void Stop() {
    if (!_is_running) return;
    _is_running = false;
    if (_wait_thread_num) _cond.NotifyAll();
    LOG(LogLevel::INFO) << "thread pool stop success";
}

但实际上上面这种做法并不推荐，太过于简单粗暴了，考虑的因素太少了。

一个线程池要退出时，应该让线程走征程的唤醒逻辑以及退出啊！

1. 如果被唤醒 && 任务队列没有任务 = 让线程退出
2. 如果被唤醒 && 任务队列有任务 = 线程不能立即退出，而应该让线程把任务处理完，在退出
3. 线程本身没有被休眠，我们应该让他把他能处理的任务全部处理完成，在退出

可以发现第 3 种情况可以归到第 2 种情况中，线程取消如下所示，为什么是这样呢？我们来看看任务执行是怎样做的。

回收线程

void Wait() {
    for (auto &t : _threads) {
        t.Join();
    }
    LOG(LogLevel::INFO) << "thread pool wait success";
}

往线程池入数据

void Enqueue(const T &t) {
    if (!_is_running) return;
    {
        LockGuard lockguard(&_lock);
        _q.push(t);
        if (_wait_thread_num > 0) _cond.NotifyOne();
    }
}