【探索Linux】—— 强大的命令行工具 P.23（线程池 —— 简单模拟）

引言

在Linux下，线程池是一种常见的并发编程模型，它能够有效地管理多个线程，提高系统的性能和资源利用率。通过线程池，可以实现多生产者多消费者模型，有效地处理并发任务，提升系统的响应速度和吞吐量。在本文中，我们将深入探讨如何在Linux环境下创建线程池，以及线程池的实现原理和使用技巧。通过深入理解线程池的概念和应用，我们可以更好地应对复杂的并发编程场景，从而提升系统的稳定性和性能表现。让我们一起探索Linux下线程池的奥秘，为并发编程的世界增添新的色彩！

一、线程池简单介绍

线程池是一种并发编程技术，用于管理和复用多个线程，以提高系统的性能和资源利用率。在线程池中，一定数量的线程被预先创建并保存在池中，当需要执行任务时，从线程池中选择一个空闲的线程来处理任务，任务执行完毕后，线程将返回到线程池中等待下一个任务。

通过使用线程池，可以避免频繁地创建和销毁线程，减少了线程创建和销毁的开销，同时也控制了并发线程的数量，避免系统资源被过度占用。线程池还可以根据系统负载情况动态调整线程数量，以更好地适应不同的工作负载。

二、Linux下线程池代码

⭕Makefile文件

thread_pool:testMain.cpp
	g++ -o $@ $^ -std=c++11 -lpthread #-DDEBUG_SHOW
clean:
	rm -f thread_pool
1
2
3
4

这个 Makefile 包含了两个规则：一个用于编译名为"testMain.cpp"的程序并生成名为"thread_pool"的可执行文件，另一个用于清理生成的可执行文件。你可以使用"make"命令编译程序，使用"make clean"命令清理生成的可执行文件。

⭕ . h 头文件

✅Task.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>

// 定义函数类型 func_t，用于表示可以接受两个整型参数并返回一个整型结果的函数
typedef std::function<int(int, int)> func_t;

// Task 类，表示一个任务
class Task
{
public:
    // 默认构造函数
    Task() {}

    // 带参数的构造函数，初始化任务的成员变量
    Task(int x, int y, func_t func) : x_(x), y_(y), func_(func)
    {}

    // 重载 () 运算符，实现任务的执行
    void operator()(const std::string &name)
    {
        // 在控制台输出任务执行的结果
        std::cout << "线程 " << name << " 处理完成, 结果是: " << x_ << "+" << y_ << "=" << func_(x_, y_) << std::endl;
    }

public:
    int x_; // 任务的参数 x
    int y_; // 任务的参数 y
    func_t func_; // 保存任务所需执行的函数
};


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
26
27
28
29
30
31
32
33
34

✅thread.hpp

#pragma once
#include <iostream>
#include <string>
#include <cstdio>

// 定义函数指针类型 fun_t，表示可以接受一个 void* 类型参数并返回一个 void* 类型结果的函数指针
typedef void *(*fun_t)(void *);

// 线程数据结构，用于保存线程的参数和名称
class ThreadData
{
public:
    void *args_; // 线程参数
    std::string name_; // 线程名称
};

// 线程类，用于创建和管理线程
class Thread
{
public:
    // 构造函数，初始化线程对象
    Thread(int num, fun_t callback, void *args) : func_(callback)
    {
        // 设置线程名称为 "Thread-num"
        char nameBuffer[64];
        snprintf(nameBuffer, sizeof nameBuffer, "Thread-%d", num);
        name_ = nameBuffer;

        // 设置线程数据的参数和名称
        tdata_.args_ = args;
        tdata_.name_ = name_;
    }

    // 启动线程
    void start()
    {
        pthread_create(&tid_, nullptr, func_, (void*)&tdata_);
    }

    // 等待线程结束
    void join()
    {
        pthread_join(tid_, nullptr);
    }

    // 获取线程名称
    std::string name()
    {
        return name_;
    }

    // 析构函数
    ~Thread()
    {
    }

private:
    std::string name_; // 线程名称
    fun_t func_;       // 线程执行的函数指针
    ThreadData tdata_; // 线程数据
    pthread_t tid_;    // 线程 ID
};


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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

✅threadPool.hpp

#pragma once
#include <iostream>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>
#include "thread.hpp"

// 互斥锁类，封装了互斥锁的操作
class Mutex
{
public:
    Mutex(pthread_mutex_t *mtx) : pmtx_(mtx)
    {
    }
    void lock()
    {
        pthread_mutex_lock(pmtx_);
    }
    void unlock()
    {
        pthread_mutex_unlock(pmtx_);
    }
    ~Mutex()
    {
    }

private:
    pthread_mutex_t *pmtx_;
};

// RAII风格的加锁方式，构造时加锁，析构时解锁
class lockGuard
{
public:
    lockGuard(pthread_mutex_t *mtx) : mtx_(mtx)
    {
        mtx_.lock();
    }
    ~lockGuard()
    {
        mtx_.unlock();
    }

private:
    Mutex mtx_;
};

// 默认线程数
const int g_thread_num = 3;

// 线程池类模板
template <class T>
class ThreadPool
{
public:
    // 获取互斥锁指针
    pthread_mutex_t *getMutex()
    {
        return &lock;
    }
    // 判断任务队列是否为空
    bool isEmpty()
    {
        return task_queue_.empty();
    }
    // 等待条件变量
    void waitCond()
    {
        pthread_cond_wait(&cond, &lock);
    }
    // 获取待执行任务
    T getTask()
    {
        T t = task_queue_.front();
        task_queue_.pop();
        return t;
    }

private:
    // 构造函数，初始化线程池
    ThreadPool(int thread_num = g_thread_num) : num_(thread_num)
    {
        pthread_mutex_init(&lock, nullptr);
        pthread_cond_init(&cond, nullptr);
        for (int i = 1; i <= num_; i++)
        {
            threads_.push_back(new Thread(i, routine, this));
        }
    }
    ThreadPool(const ThreadPool<T> &other) = delete;
    const ThreadPool<T> &operator=(const ThreadPool<T> &other) = delete;

public:
    // 获取线程池单例
    static ThreadPool<T> *getThreadPool(int num = g_thread_num)
    {
        if (nullptr == thread_ptr)
        {
            lockGuard lockguard(&mutex);
            if (nullptr == thread_ptr)
            {
                thread_ptr = new ThreadPool<T>(num);
            }
        }
        return thread_ptr;
    }
    // 启动线程池中的线程
    void run()
    {
        for (auto &iter : threads_)
        {
            iter->start();
            std::cout << iter->name() << " 启动成功" << std::endl;
        }
    }
    // 线程执行的函数
    static void *routine(void *args)
    {
        ThreadData *td = (ThreadData *)args;
        ThreadPool<T> *tp = (ThreadPool<T> *)td->args_;
        while (true)
        {
            T task;
            {
                lockGuard lockguard(tp->getMutex());
                while (tp->isEmpty())
                    tp->waitCond();
                task = tp->getTask();
            }
            task(td->name_);
        }
    }
    // 向任务队列中添加任务
    void pushTask(const T &task)
    {
        lockGuard lockguard(&lock);
        task_queue_.push(task);
        pthread_cond_signal(&cond);
    }
    // 析构函数，销毁线程池
    ~ThreadPool()
    {
        for (auto &iter : threads_)
        {
            iter->join();
            delete iter;
        }
        pthread_mutex_destroy(&lock);
        pthread_cond_destroy(&cond);
    }

private:
    std::vector<Thread *> threads_; // 线程对象数组
    int num_; // 线程数量
    std::queue<T> task_queue_; // 任务队列

    static ThreadPool<T> *thread_ptr; // 线程池单例指针
    static pthread_mutex_t mutex; // 线程池单例的互斥锁

    pthread_mutex_t lock; // 线程池内部使用的互斥锁
    pthread_cond_t cond; // 线程池内部使用的条件变量
};

// 静态成员初始化
template <typename T>
ThreadPool<T> *ThreadPool<T>::thread_ptr = nullptr;

template <typename T>
pthread_mutex_t ThreadPool<T>::mutex = PTHREAD_MUTEX_INITIALIZER;

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

⭕ . cpp 文件

✅testMain.cpp

#include "threadPool.hpp" // 包含线程池头文件
#include "Task.hpp" // 包含任务类的头文件
#include <ctime>
#include <cstdlib>
#include <iostream>
#include <unistd.h>

int main()
{
    srand((unsigned long)time(nullptr) ^ getpid()); // 初始化随机数种子

    // 获取线程池单例并运行线程
    ThreadPool<Task>::getThreadPool()->run();

    while(true)
    {
        // 生产任务的过程，制作任务的时候，要花时间
        int x = rand() % 100 + 1; // 随机生成一个范围在1到100之间的整数x
        usleep(7721); // 模拟制作任务需要的时间
        int y = rand() % 30 + 1; // 随机生成一个范围在1到30之间的整数y
        Task t(x, y, [](int x, int y) -> int { // 创建任务对象t，执行加法操作
            return x + y;
        });

        std::cout << "制作任务完成: " << x << "+" << y << "=?" << std::endl; // 输出任务信息

        // 推送任务到线程池中
        ThreadPool<Task>::getThreadPool()->pushTask(t);

        sleep(1); // 暂停一秒钟
    }
    return 0;
}
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
26
27
28
29
30
31
32
33

这段代码主要实现了一个简单的任务生产者，不断地生成任务并将任务推送到线程池中执行。

三、线程池的优点

线程池能够有效地管理线程，提高系统的性能和响应速度，同时简化了线程管理的复杂性，是多线程编程中常用的一种技术。下面是它的优点

1. 提高性能：线程池可以减少线程创建和销毁的开销，通过重用线程，避免了频繁地创建和销毁线程所带来的性能损耗。

2. 控制并发度：线程池可以限制同时执行的线程数量，从而控制系统的并发度，防止因为过多线程导致系统资源被耗尽。

3. 提高响应速度：由于线程池中的线程已经创建好并处于就绪状态，当任务到达时可以立即执行，从而减少了任务等待执行的时间，提高了系统的响应速度。

4. 简化线程管理：线程池封装了线程的创建、销毁、调度等操作，简化了线程管理的复杂性，提高了代码的可维护性。

5. 控制资源占用：线程池可以限制系统中同时存在的线程数量，从而控制系统对资源（如内存、CPU）的占用，防止资源被耗尽导致系统崩溃。

总而言之，线程池提供了一种高效、可控的线程管理机制，适用于处理大量并发任务的场景，是提高系统性能和响应速度的重要工具之一。

温馨提示

感谢您对博主文章的关注与支持！如果您喜欢这篇文章，可以点赞、评论和分享给您的同学，这将对我提供巨大的鼓励和支持。另外，我计划在未来的更新中持续探讨与本文相关的内容。我会为您带来更多关于Linux以及C++编程技术问题的深入解析、应用案例和趣味玩法等。如果感兴趣的话可以关注博主的更新，不要错过任何精彩内容！

再次感谢您的支持和关注。我们期待与您建立更紧密的互动，共同探索Linux、C++、算法和编程的奥秘。祝您生活愉快，排便顺畅！