慕雪的小助手正在绞尽脑汁···
慕雪小助手的总结
LongCat-Flash-Chat

今天来写一个简单版本的线程池

1.啥是线程池

池塘，顾名思义，线程池就是一个有很多线程的容器。

我们只需要把任务交到这个线程的池子里面，其就能帮我们多线程执行任务，计算出结果。

与阻塞队列不同的是，线程池中内有一个队列用于任务管理，并帮我们封装了线程创建的工作。我们不再需要在主执行流里面创建线程（创建线程也是有时间消耗的），而是只关注于任务的创建，交给线程池来运行并产生结果就OK了

前面已经学习过阻塞队列了，此时再来写线程池，就没有那么困难了！

本次线程池的设计还会采用单例模式，同一个模板类型的任务，只需要一个线程池即可

1.1 简单复习单例模式

单例模式分为两种设计方式，一个是懒汉，一个是饿汉

  • 懒汉：刚开始先不创建单例，等第一次使用的时候在创建；缺点是第一次获取单例需要等待，优点是程序启动快
  • 饿汉：main函数执行前，就将单例创建起来；缺点是程序启动会比较慢，优点是启动之后获取单例会快

2.代码示例-处理task

2.1 成员变量

因为是线程池，需要在内部创建出线程来运行，所以我们需要一个num来标识需要创建的线程的数量

1
2
3
4
5
6
7
8
9
10
template <class T>
class ThreadPool{
private:
    bool _isStart;  // 线程池子是否启动
    int _threadNum; // 线程数量
    queue<T> _tq;   // 任务队列
    pthread_mutex_t _mutex;// 锁
    pthread_cond_t _cond;  // 条件变量
    static ThreadPool<T> *instance; // 单例模式需要用到的指针
}

这里我们并不需要弄一个数组来存放已经创建的线程，因为我们并不关心线程的退出信息，也不需要对线程进行管理。在创建好线程之后，直接detach即可

static变量我们需要在类外初始化，因为是模板类型，所以还需要带上template关键字

1
2
3
// 初始化static变量
template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;

2.2 构造/析构（单例）

本次使用的是懒汉模式的单例，提供一个指针作为单例，不开放构造函数（构造函数私有化）

同时，利用delete关键字，禁止拷贝构造和赋值重载；析构依旧保持公有

1
2
3
4
5
6
7
8
9
10
11
private:
    ThreadPool(int num = DEFALUT_NUM)
        : _threadNum(num),
          _isStart(false)
    {
        assert(num > 0);
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
    }
    ThreadPool(const ThreadPool<T> &) = delete;//取消拷贝
    void operator=(const ThreadPool<T> &) = delete;//取消赋值

析构函数并不需要进行过多处理，将锁和条件变量销毁即可

1
2
3
4
5
~ThreadPool()
{
    pthread_mutex_destroy(&_mutex);
    pthread_cond_destroy(&_cond);
}

这种情况下，我们还需要有一个static成员函数来获取单例；在之前的单例模式博客中，提到当初实现的懒汉模式是线程不安全的，因为没有对线程进行加锁，避免多个执行流同时获取单例，导致单例对象冲突的问题。

现在学习了linux的加锁操作，就可以避免掉这个bug了

两次nullptr判断

其中关于两次nullptr判断的原因，详见注释

  • 第一个判断是为了保证单例，只要单例存在了，就不再创建单例
  • 第二个判断是保证线程安全，可能会出现线程a在创建单例，线程b在锁中等待的情况；此时如果不进行第二次nullptr判断，线程b从锁中被唤醒后，又会继续执行，多创建了一个单例！
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public:
    static ThreadPool<T> *getInstance()
    {
        static pthread_mutex_t mt;//使用static，锁只会创建一次；避免多次实例化，一个执行流一个锁就失去效果了
        pthread_mutex_init(&mt,nullptr);
        if (instance == nullptr) // 第一次判断
        {
            pthread_mutex_lock(&mt);// 加锁，保证只有一个执行流走到这里
            if (instance == nullptr)// 第二次判断是来确认的，避免出现在加锁前，被其他执行流获取过实例了
            {
                instance = new ThreadPool<T>();// 确认是null，创建单例
            }
        }

        pthread_mutex_unlock(&mt);
        pthread_mutex_destroy(&mt);
        return instance;
    }

2.3 启动线程池

有了线程池，接下来要做的就是启动它😁

启动之前，我们需要assert判断一下该线程池是否已经启动了，避免多次启动线程池出现问题。启动完成之后，更新isStart的状态值

1
2
3
4
5
6
7
8
9
10
11
12
void start()
{
    assert(!_isStart);//如果开启了，那么就不能执行该函数
    for (int i = 0; i < _threadNum; i++)
    {
        pthread_t temp;
        pthread_create(&temp, nullptr, threadRoutine, this);//把this当参数传入
        usleep(100);
        pthread_detach(temp);//分离线程
    }
    _isStart = true;//标识状态，代表线程池已经启动了
}

这里还有另外一个函数threadRoutine，这是每一个线程需要执行的函数，其为static函数。这里我们获取到的都是单例的this指针，访问成员都需要通过this指针来访问

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
static void *threadRoutine(void *args)
{
    ThreadPool<T> *tp = static_cast<ThreadPool<T> *>(args);//c++强转
    while (1)
    {
        tp->lockQueue();
        while (!tp->haveTask())
        {
            tp->waitForTask();
        }
        // 任务被拿到了线程的上下文中
        T t = tp->pop();
        tp->unlockQueue();

        // 规定每一个封装的task对象都需要有一个run函数
        t.resultPrint(t.run());//运行并打印结果
    }
}

2.4 封装的加锁/解锁/通知操作

这部分操作比较简单，就不多提了。其实就是把已有的函数改个名字，变成无参可直接调用的函数罢了。

1
2
3
4
5
6
7
8
9
10
11
12
private:
    void lockQueue() { pthread_mutex_lock(&_mutex); }
    void unlockQueue() { pthread_mutex_unlock(&_mutex); }
    bool haveTask() { return !_tq.empty(); }
    void waitForTask() { pthread_cond_wait(&_cond, &_mutex); }
    void singalThread() { pthread_cond_signal(&_cond); }
    T pop()
    {
        T temp = _tq.front();
        _tq.pop();
        return temp;
    }

其中pop()函数设置为了私有，因为线程池会自己开始处理任务，所以不需要外部pop

2.5 插入任务

最后就只剩下任务的插入了，插入一个任务后，使用条件变量，唤醒线程池中的一个线程来执行这个任务！

1
2
3
4
5
6
7
8
//往线程池中给任务
void push(const T &in)
{
    lockQueue();
    _tq.push(in);//插入任务
    singalThread();//任务插入后，唤醒一个线程来执行
    unlockQueue();
}

到这里，线程池就大功告成了！

3.测试

本次测试依旧使用了在线程博客中提到过的task.hpp，完整代码详见我的gitee仓库

因为使用了线程池，主执行流只需要来派发任务即可；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#include "threadpool.hpp"
#include "task.hpp"
#include <string>
#include <time.h>

int main()
{
    const string operators = "+/*/%";
    ThreadPool<Task>*tp = ThreadPool<Task>::getInstance();
    tp->start();

    srand((unsigned long)time(nullptr) ^ getpid() ^ pthread_self());
    // 派发任务的线程
    while(1)
    {
        int one = rand()%50;
        int two = rand()%10;
        char oper = operators[rand()%operators.size()];
        cout << "[" << pthread_self() << "] 主线程派发计算任务: " << one << oper << two << "=?" << "\n";
        Task t(one, two, oper);
        tp->push(t);
        sleep(1);
    }
    
}

此时线程池就会帮我们运行，并将结果输出！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
[muxue@bt-7274:~/git/linux/code/23-01-18 threadpool]$ ./test
[140202992179008] 主线程派发计算任务: 14/8=?
[140202973767424] 新线程完成计算任务: 14/8=1
[140202992179008] 主线程派发计算任务: 43*2=?
[140202965374720] 新线程完成计算任务: 43*2=86
[140202992179008] 主线程派发计算任务: 10/9=?
[140202956982016] 新线程完成计算任务: 10/9=1
[140202992179008] 主线程派发计算任务: 25*9=?
[140202948589312] 新线程完成计算任务: 25*9=225
[140202992179008] 主线程派发计算任务: 8/0=?
div zero, abort
[140202940196608] 新线程完成计算任务: 8/0=-1
[140202992179008] 主线程派发计算任务: 38%1=?
[140202973767424] 新线程完成计算任务: 38%1=0
[140202992179008] 主线程派发计算任务: 23/7=?
[140202965374720] 新线程完成计算任务: 23/7=3
[140202992179008] 主线程派发计算任务: 4%4=?
[140202956982016] 新线程完成计算任务: 4%4=0
[140202992179008] 主线程派发计算任务: 44*8=?
[140202948589312] 新线程完成计算任务: 44*8=352
[140202992179008] 主线程派发计算任务: 4/2=?

3.1 修改轻量级进程的名字

Linux提供了一个有趣的接口，可以允许我们修改轻量级进程的名字；

没有修改的时候，默认的名字都是该进程的可执行程序的名字

1
2
3
4
5
6
7
8
9
[muxue@bt-7274:~/git/linux/code/23-01-18 threadpool]$ ps -aL
  PID   LWP TTY          TIME CMD
 6592  6592 pts/7    00:00:00 test
 6592  6593 pts/7    00:00:00 test
 6592  6594 pts/7    00:00:00 test
 6592  6595 pts/7    00:00:00 test
 6592  6596 pts/7    00:00:00 test
 6592  6597 pts/7    00:00:00 test
 6730  6730 pts/8    00:00:00 ps

我们使用prctl接口，修改名字；这个接口的作用是对一个进程进行操作。

1
2
3
#include <sys/prctl.h>
int prctl(int option, unsigned long arg2, unsigned long arg3,
        unsigned long arg4, unsigned long arg5);

其中修改线程名字的操作如下

1
prctl(PR_SET_NAME, "handler");//修改线程名字为handler

分别修改主执行流和线程池中线程的名字，即可获得不一样的结果

1
2
3
4
5
6
7
8
9
[muxue@bt-7274:~/git/linux/code/23-01-18 threadpool]$ ps -aL
  PID   LWP TTY          TIME CMD
 7793  7793 pts/7    00:00:00 master
 7793  7794 pts/7    00:00:00 handler
 7793  7795 pts/7    00:00:00 handler
 7793  7796 pts/7    00:00:00 handler
 7793  7797 pts/7    00:00:00 handler
 7793  7798 pts/7    00:00:00 handler
 7828  7828 pts/8    00:00:00 ps

这样可以用于标识线程的属性，还是有些用的！

The end

本篇博客到这里就over啦，有啥问题欢迎评论区提出哦！

QQ图片20220413084241