【Linux练习生】线程池

• 本节知识所需代码已同步到gitee-> https://gitee.com/ZMZZZhao/linux-git/tree/master/thread_pool
• 本文收录于专栏：Linux
• 关注作者，持续阅读作者的文章，学习更多知识！
  https://blog.csdn.net/weixin_53306029?spm=1001.2014.3001.5343

一、线程池的简介

• 通常我们使用多线程的方式是，需要时创建一个新的线程，在这个线程里执行特定的任务，然后在任务完成后退出。这在一般的应用里已经能够满足我们应用的需求，毕竟我们并不是什么时候都需要创建大量的线程，并在它们执行一个简单的任务后销毁。

• 但是在一些web、email、database等应用里，比如彩铃，我们的应用在任何时候都要准备应对数目巨大的连接请求，同时，这些请求所要完成的任务却又可能非常的简单，即只占用很少的处理时间。这时，我们的应用有可能处于不停的创建线程并销毁线程的状态。虽说比起进程的创建，线程的创建时间已经大大缩短，但是如果需要频繁的创建线程，并且每个线程所占用的处理时间又非常简短，则线程创建和销毁带给处理器的额外负担就变得比较大了。

• 线程池的作用正是在这种情况下有效的降低频繁创建销毁线程所带来的额外开销。一般来说，线程池都是采用预创建的技术，在应用启动之初便预先创建一定数目的线程。应用在运行的过程中，需要时可以从这些线程所组成的线程池里申请分配一个空闲的线程，来执行一定的任务，任务完成后，并不是将线程销毁，而是将它返还给线程池，由线程池自行管理。如果线程池中预先分配的线程已经全部分配完毕，但此时又有新的任务请求，则线程池会动态的创建新的线程去适应这个请求。当然，有可能，某些时段应用并不需要执行很多的任务，导致了线程池中的线程大多处于空闲的状态，为了节省系统资源，线程池就需要动态的销毁其中的一部分空闲线程。因此，线程池都需要一个管理者，按照一定的要求去动态的维护其中线程的数目。

• 基于上面的技术，线程池将频繁创建和销毁线程所带来的开销分摊到了每个具体执行的任务上，执行的次数越多，则分摊到每个任务上的开销就越小。

• 当然，如果线程创建销毁所带来的开销与线程执行任务的开销相比微不足道，可以忽略不计，则线程池并没有使用的必要。比如，FTP、Telnet等应用时。

二、线程池的优点

• 线程池避免了在处理短时间任务时创建与销毁线程的代价。
• 线程池不仅能够保证内核充分利用，还能防止过分调度。

注意： 线程池中可用线程的数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。

三、线程池的应用场景

线程池常见的应用场景如下：

• 需要大量的线程来完成任务，且完成任务的时间比较短。
• 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
• 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。

相关解释：

• 像Web服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。
• 对于长时间的任务，比如Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
• 突发性大量客户请求，在没有线程池的情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，但短时间内产生大量线程可能使内存到达极限，出现错误。

四、模拟实现一个线程池

下面我们实现一个简单的线程池，线程池中提供了一个任务队列，以及若干个线程（多线程）。

1. 线程池实现原理

• 线程池通过一个线程安全的阻塞任务队列加上一个或一个以上的线程实现。

• 线程池中的线程可以从阻塞任务队列中获取任务然后进行任务处理。

• 当线程都处于繁忙状态时可以将任务加入阻塞队列中，等到其它的线程空闲后进行处理。

2. 线程池基本框架
线程池的主体是一个任务队列和n个线程：

• 任务队列用来保存外部传入的需要解决的任务。
• 而线程池中的n个线程负责从任务队列中拿出任务并解决。

PS：类型模板参数T由我们创建线程池对象时显示传入，它代表我们要处理的任务的类型。

线程池类实现

#pragma once

#include
#include
#include
#include
#include
using namespace std;

namespace ns_threadpool
{
	// 线程池中线程个数的缺省值
    const int g_num=5;
    template<class T>
    class ThreadPool
    {
    private:
        int num_;
        queue<T> task_queue_;//任务队列(该成员是临界资源)

        pthread_mutex_t mtx_;
        pthread_cond_t cond_; 

    public:
        void Lock()
        {
            pthread_mutex_lock(&mtx_);
        }

        void Unlock()
        {
            pthread_mutex_unlock(&mtx_);
        }

        void Wait()
        {
            pthread_cond_wait(&cond_,&mtx_);
        }

        void Wakeup()
        {
            pthread_cond_signal(&cond_);
        }

        bool IsEmpty()
        {
            return task_queue_.empty();
        }
    public:
        ThreadPool(int num=g_num):num_(num)
        {
            pthread_mutex_init(&mtx_,nullptr);
            pthread_cond_init(&cond_,nullptr);
        }

        //在类中，要让线程执行类内成员方法，是不可行的
        //所以必须让线程执行静态方法(在类内设为静态方法后，默认是没有this指针的，并且不能直接访问类内的非static成员)
        //线程池中线程的执行例程
        static void* Rountine(void* args)//注意要设为静态方法
        {
            pthread_detach(pthread_self());//分离线程

            ThreadPool<T> *tp=(ThreadPool<T> *)args;
            //不断从任务队列获取任务进行处理
            while(true)
            {
                tp->Lock();
                while(tp->IsEmpty())
                {
                    //任务队列为空，线程该做什么呢
                    tp->Wait();
                }
                //任务队列有任务了
                T t;
                tp->PopTask(&t);
                tp->Unlock();

                t.Run();
            }
        }

        void InitThreadPool()
        {
            pthread_t tid;
            for(int i=0;i<num_;i++)
            {
                pthread_create(&tid,nullptr,Rountine,(void*)this);//注意参数传入this指针
            }
        }
		//往任务队列塞任务（主线程调用）
        void PushTask(const T& in)
        {
            Lock();
            task_queue_.push(in);
            Unlock();
            Wakeup();
        }
	    //从任务队列获取任务（线程池中的线程调用）
        void PopTask(T *out)
        {
            *out=task_queue_.front();
            task_queue_.pop();
        }

        ~ThreadPool()
        {
            pthread_mutex_destroy(&mtx_);
            pthread_cond_destroy(&cond_);
        }

    };
}
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

注意：

• 当某线程被唤醒时，其可能是被异常或是伪唤醒，或者是一些广播类的唤醒线程操作而导致所有线程被唤醒，使得在被唤醒的若干线程中，只有个别线程能拿到任务。此时应该让被唤醒的线程再次判断是否满足被唤醒条件，所以在判断任务队列是否为空时，应该使用while进行判断，而不是if。
• pthread_cond_broadcast函数的作用是唤醒条件变量下的所有线程，而外部可能只Push了一个任务，我们却把全部在等待的线程都唤醒了，此时这些线程就都会去任务队列获取任务，但最终只有一个线程能得到任务。一瞬间唤醒大量的线程可能会导致系统震荡，这叫做惊群效应。因此在唤醒线程时最好使用pthread_cond_signal函数唤醒正在等待的一个线程即可。
• 当线程从任务队列中拿到任务后，该任务就已经属于当前线程了，与其他线程已经没有关系了，因此应该在解锁之后再进行处理任务，而不是在解锁之前进行。因为处理任务的过程可能会耗费一定的时间，所以我们不要将其放到临界区当中。
• 如果将处理任务的过程放到临界区当中，那么当某一线程从任务队列拿到任务后，其他线程还需要等待该线程将任务处理完后，才有机会进入临界区。此时虽然是线程池，但最终我们可能并没有让多线程并行的执行起来。

为什么线程池中需要有互斥锁和条件变量？

线程池中的任务队列是会被多个执行流同时访问的临界资源，因此我们需要引入互斥锁对任务队列进行保护。

线程池当中的线程要从任务队列里拿任务，前提条件是任务队列中必须要有任务，因此线程池当中的线程在拿任务之前，需要先判断任务队列当中是否有任务，若此时任务队列为空，那么该线程应该进行等待，直到任务队列中有任务时再将其唤醒，因此我们需要引入条件变量。

当外部线程向任务队列中Push一个任务后，此时可能有线程正处于等待状态，因此在新增任务后需要唤醒在条件变量下等待的线程。

为什么线程池中的线程执行例程需要设置为静态方法？

使用pthread_create函数创建线程时，需要为创建的线程传入一个Routine（执行例程），该Routine只有一个参数类型为void的参数，以及返回类型为void的返回值。

而此时Routine作为类的成员函数，该函数的第一个参数是隐藏的this指针，因此这里的Routine函数，虽然看起来只有一个参数，而实际上它有两个参数，此时直接将该Routine函数作为创建线程时的执行例程是不行的，无法通过编译。

静态成员函数属于类，而不属于某个对象，也就是说静态成员函数是没有隐藏的this指针的，因此我们需要将Routine设置为静态方法，此时Routine函数才真正只有一个参数类型为void*的参数。

但是在静态成员函数内部无法调用非静态成员函数，而我们需要在Routine函数当中调用该类的某些非静态成员函数，比如Pop。因此我们需要在创建线程时，向Routine函数传入的当前对象的this指针，此时我们就能够通过该this指针在Routine函数内部调用非静态成员函数了。

任务类的设计

• 我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的。
• 每个线程池对象可以解决一种类型的任务，这个任务的相关数据和解决逻辑需要我们自己写，解决逻辑部分需要专门写到类的公有成员函数Task::Run()中，这样线程池中的线程在拿到任务后就可以统一执行该函数来解决任务。

下面我们自己写一个任务类，它的功能是进行正整数的加减乘除、取模运算：

#pragma once

#include 
#include 

namespace ns_task
{
    class Task
    {
    private:
        int x_;
        int y_;
        char op_; //+/*/%
    public:
        // void (*callback)();
        Task() {}
        Task(int x, int y, char op) : x_(x), y_(y), op_(op)
        {
        }
        std::string Show()
        {
            std::string message = std::to_string(x_);
            message += op_;
            message += std::to_string(y_);
            message += "=?";
            return message;
        }
        // 解决任务的Run函数
        int Run()
        {
            int res = 0;
            switch (op_)
            {
            case '+':
                res = x_ + y_;
                break;
            case '-':
                res = x_ - y_;
                break;
            case '*':
                res = x_ * y_;
                break;
            case '/':
                res = x_ / y_;
                break;
            case '%':
                res = x_ % y_;
                break;
            default:
                std::cout << "bug??" << std::endl;
                break;
            }
            std::cout << "当前任务正在被: " << pthread_self() << " 处理: " \
            << x_ << op_ << y_ << "=" << res << std::endl;
            return res;
        }
        int operator()()
        {
            return Run();
        }
        ~Task() {}
    };
}
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

测试线程池

结合上面声明的线程池类和任务类，接下来对我们写的线程池进行测试。在主线程中执行以下逻辑：

主线程就负责不断向任务队列当中Push任务就行了，此后线程池当中的线程会从任务队列当中获取到这些任务并进行处理。

• 创建一个任务类型为class Task的线程池对象。
• 通过种子随机生成不同的运算数和运算符来构造Task对象。
• 把Task对象不断地插入到任务队列中，让线程池中的线程去解决任务。

#include"thread_pool.hpp"
#include"Task.hpp"

#include
#include
using namespace ns_threadpool;
using namespace ns_task;
int main()
{
    ThreadPool<Task> *tp=new ThreadPool<Task>(10);//创建线程池（并创建10个线程）
    tp->InitThreadPool();//初始化线程池当中的线程

    srand((long long)time(nullptr));
    //不断往任务队列塞计算任务
    while(true)
    {
         sleep(1);
         Task t(rand()%20+1,rand()%10+1,"+-*/%"[rand()%5]);
         tp->PushTask(t);
    }

    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

贴士：运行代码后一瞬间会有11个线程，其中1个是主线程，另外10个是线程池内处理任务的线程。

运行代码后我们会发现：这10个线程在处理时会呈现出一定的顺序性，因为主线程是每秒Push一个任务，这10个线程只会有一个线程获取到该任务，其他线程都会在等待队列中进行等待，当该线程处理完任务后就会因为任务队列为空而排到等待队列的最后，当主线程再次Push一个任务后会唤醒等待队列首部的一个线程，这个线程处理完任务后又会排到等待队列的最后。

---

– the End –

以上就是我分享的Linux线程池相关内容，感谢阅读！