Linux多线程---线程池实现

线程池的定义

线程池（Thread Pool）是一种多线程处理模式，用于管理和复用线程资源，避免频繁创建和销毁线程带来的性能开销。其核心思想是预先创建一定数量的线程，当有任务提交时，从线程池中获取可用线程执行任务，任务完成后线程不销毁而是返回线程池等待下一个任务。

看完定义，你可能会觉得线程池和进程池好像很像，他们都是管理并发执行单元的技术，但是两者在实现机制，资源消耗和使用场景方面有很大差异。

首先，线程池内的线程是共享进程地址空间的，线程切换开销很小，而进程池内的进程都有自己的独立空间，切换开销很大。

同时，由于线程间共享进程地址空间，所以线程间通信就无需借助IPC，但是需要锁和条件变量控制线程之间的互斥和同步，线程池内的线程资源隔离性也因此比较差，一个线程崩溃会影响整个进程。

线程池适用场景

• I/O 密集型任务（如网络请求、文件读写）：线程阻塞时不影响其他线程执行。
• 需要共享内存的场景（如缓存数据共享）。
• 任务执行时间短、频繁创建销毁线程的场景（如 Web 服务器请求处理）。

线程池的实现

 主线程将任务push进入任务队列，其他多线程用于从任务队列获取任务并执行。

代码：

首先我们需要一个ThreadPool类来管理线程池就像进程池有ProcessPool一样。

先描述再组织，线程池中有线程数量、多线程、任务队列等基本内容，任务队列通过queue管理，线程间互斥和同步需要锁和条件变量。

#pragma once

#include <vector>
#include <queue>
#include <pthread.h>
#include <string>

#define NUM 5//默认线程池内线程数

template<typename T>
class ThreadPool
{
public:
    ThreadPool(int num=NUM):_threadnum(num)
    {
        pthread_mutex_init(&_mutex,nullptr);
        pthread_cond_init(&cond,nullptr);
    }

    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&cond);
    }

private:
    int _threadnum;
    queue<T> _tasklist;

    pthread_mutex_t _mutex;//锁和条件变量保证线程间的互斥同步
    pthread_cond_t _cond;


};

 完成基础定义，后面还需要完成：

1. 线程池的初始化
2. 线程执行的函数(执行任务)
3. 任务列表push

ThreadPool.hpp:

#pragma once

#include <vector>
#include<iostream>
#include <queue>
#include <pthread.h>
#include <string>

#define NUM 5 // 默认线程池内线程数

template <typename T>
class ThreadPool
{
private:
    void Lock()
    {
        pthread_mutex_lock(&_mutex);
    }
    void Unlock()
    {
        pthread_mutex_unlock(&_mutex);
    }
    void Wait()
    {
        pthread_cond_wait(&_cond, &_mutex);
    }
    void Wakeup()
    {
        pthread_cond_signal(&_cond);
    }

public:
    ThreadPool(int num = NUM) : _threadnum(num)
    {
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
    }

    static void *threadroutine(void *arg)
    {
        pthread_detach(pthread_self());//线程分离无需join
        ThreadPool *self = static_cast<ThreadPool *>(arg);
        while (true)
        {
            self->Lock();
            while (self->_tasklist.empty())//任务列表空则等待
            {
                self->Wait();
            }
            T task = self->_tasklist.front(); 
            self->_tasklist.pop();
            self->Unlock();  
            task();
        }
    }

    void InitThreadpool()
    {
        for (int i = 0; i < _threadnum; i++)
        {
            pthread_t tid;
            pthread_create(&tid, nullptr, threadroutine, this);
            std::cout<<"thread create success..."<<std::endl;
        }
    }
    void push(const T& task)
    {
        Lock();
        _tasklist.push(task);
        Wakeup();//唤醒条件变量通知进程可以获取任务了
        Unlock();
    }

    

    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }

private:
    int _threadnum;
    std::queue<T> _tasklist;

    pthread_mutex_t _mutex; // 锁和条件变量保证线程间的互斥同步
    pthread_cond_t _cond;
};

为什么线程中需要锁和条件变量？

为了控制线程间的互斥和同步，线程中的任务队列是会被多给线程访问的临界资源，为了保护临界资源我们使用锁。

线程从任务队列中取任务的前提条件是任务队列中有任务，如果执行任务的线程不停的拿锁访问任务队列，同时任务队列中又没有任务可执行，会导致主线程无法拿到锁无法放置任务进而导致效率低下。所以任务队列中没有任务的时候，设置条件变量让线程阻塞并且把拿到的锁释放出来，等待任务队列中有任务出现，然后被唤醒。

注意：

• 当某线程被唤醒时，其可能是被异常或是伪唤醒，或者是一些广播类的唤醒线程操作而导致所有线程被唤醒，使得在被唤醒的若干线程中，只有个别线程能拿到任务。此时应该让被唤醒的线程再次判断是否满足被唤醒条件，所以在判断任务队列是否为空时，应该使用while进行判断，而不是if。
• 任务的执行是属于每一个线程自己的，所以不需要用锁保护起来，并且任务的执行时间如果很长会耽误其他线程拿到任务，导致线程池不是并发的。

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

因为threadroutine方法有必须的格式，返回void*类型参数类型也只能是void*，然后类方法会自动带上this指针，导致不符合格式，static修饰的类方法不会带this指针，但是我们要用到this指针怎么办呢，那就再pthread _create的时候将this指针传入进去，使用的时候强转就行。

任务类型的设计

我们将线程池进行了模板化，因此线程池当中存储的任务类型可以是任意的，但无论该任务是什么类型的，在该任务类当中都必须包含一个Run方法，当我们处理该类型的任务时只需调用该Run方法即可。

task.hpp:

#pragma once

#include <iostream>

//任务类
class Task
{
public:
	Task(int x = 0, int y = 0, char op = 0)
		: _x(x), _y(y), _op(op)
	{}
	~Task()
	{}

	//处理任务的方法
	void Run()
	{
		int result = 0;
		switch (_op)
		{
		case '+':
			result = _x + _y;
			break;
		case '-':
			result = _x - _y;
			break;
		case '*':
			result = _x * _y;
			break;
		case '/':
			if (_y == 0){
				std::cerr << "Error: div zero!" << std::endl;
				return;
			}
			else{
				result = _x / _y;
			}
			break;
		case '%':
			if (_y == 0){
				std::cerr << "Error: mod zero!" << std::endl;
				return;
			}
			else{
				result = _x % _y;
			}
			break;
		default:
			std::cerr << "operation error!" << std::endl;
			return;
		}
		std::cout << "thread[" << pthread_self() << "]:" << _x << _op << _y << "=" << result << std::endl;
	}
    void operator()()
    {
        Run();
    }
    
private:
	int _x;
	int _y;
	char _op;
};

主线程的作用就是创建好线程池后不断地像任务队列中push任务 。

 main.cc:

#include<iostream>
#include"ThreadPool.hpp"
#include"task.hpp"
#include<unistd.h>


int main()
{
    ThreadPool<Task>* tp=new ThreadPool<Task>(5);
    tp->InitThreadpool();

    const char* op = "+-*/%";
    while(true)
    {
        int a=rand()%10+1;
        int b=rand()%5+2;
        int index = rand() % 5;
        Task t(a,b,op[index]);
        tp->push(t);
        sleep(1);
        
    }


    return 0;
}

执行结果：