Linux——线程池

目录

线程池的概念

线程池的优点

线程池的实现

【注意】

线程池的线程安全

日志文件的实现

---

线程池的概念

        线程池也是一种池化技术，可以预先申请一批线程，当我们后续有任务的时候就可以直接用，这本质上是一种空间换时间的策略。

        如果有任务来的时候再创建线程，那成本又要提高，又要初始化，又要创建数据结构。

线程池的优点

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

线程池的实现

        我们这次要实现的线程池就是这样，让主线程派发任务，让线程池中的线程处理任务，这也是一个生产者消费者模型。

// thread.hpp
// 把线程封装一下
#pragma once#include <iostream>
#include <string>
#include <cstdio>
#include <vector>
#include <queue>
#include <unistd.h>using namespace std;typedef void*(*func_t)(void*);class ThreadData
{
public:string name_;void* args_;
};class Thread
{
public:Thread(int num, func_t callback, void* args):func_(callback){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);}string name(){return name_;}~Thread(){}
private:string name_;pthread_t tid_;ThreadData tdata_;func_t func_;
};

// threadPool.hpp#pragma once#include "thread.hpp"
#include "lockGuard.hpp"
#include "log.hpp"const int g_default_num = 3;template <class T>
class ThreadPool
{
public:// 通过接口获得成员变量pthread_mutex_t* getMutex(){return &lock_;}void waitCond(){pthread_cond_wait(&cond_, &lock_);}bool isEmpty(){return task_queue_.empty();}
public:ThreadPool(int thread_num = g_default_num) // 初始化后，就已经有了对象，也有了this指针:num_(thread_num){pthread_mutex_init(&lock_, nullptr);pthread_cond_init(&cond_, nullptr);for (int i = 0; i < num_; i++){threads_.push_back(new Thread(i + 1, routine, this) ); // 通过传入this指针就可以拿到ThreadPool中的task_queue}}void run(){for (auto& iter : threads_){iter->start();cout << iter->name() << "启动成功" << endl;}}// 去掉this指针// 消费的过程static void* routine(void* args){ThreadData* td = (ThreadData*)args;ThreadPool<T>* tq = (ThreadPool<T>*)td->args_; // 去掉this指针就无法访问成员方法了，通过创建线程的时候传入this拿到线程池对象while (true){T task;{lockGuard lockguard(tq->getMutex());  // 加锁while (tq->isEmpty()) tq->waitCond(); // 检测// 读取任务task = tq->getTask();}// 仿函数cout << td->name_ << ", 消费者：" << task._x << " + " << task._y << " = " << task() << endl;// sleep(1);}}void pushTask(const T& task){lockGuard lockguard(&lock_);task_queue_.push(task);pthread_cond_signal(&cond_);}T getTask(){T t = task_queue_.front();task_queue_.pop();return t;}void joins(){for (auto& iter : threads_){iter->join();}}~ThreadPool(){for (auto& iter : threads_){delete iter;}pthread_mutex_destroy(&lock_);pthread_cond_destroy(&cond_);}
private:vector<Thread*> threads_;int num_;queue<T> task_queue_;  // 任务队列pthread_mutex_t lock_; // 互斥锁pthread_cond_t cond_;  // 条件变量
};

// testMain.cc
#include "threadPool.hpp"
#include "Task.hpp"
#include <ctime>int Add(int x, int y)
{return x + y;
}int main()
{srand((unsigned)time(nullptr));cout << "hello thread pool" << endl;ThreadPool<Task> *tp = new ThreadPool<Task>();tp->run();while (true){int x = rand() % 10 + 1;usleep(rand() % 1000);int y = rand() % 10 + 1;Task t(x, y, Add);tp->pushTask(t);cout << "生产者：" << x << " + " << y << " = ? " << endl;//sleep(1);}tp->joins();return 0;
}

【注意】

1. 线程池中的任务队列会被多个执行流访问，因此我们需要互斥锁对任务队列进行保护。
2. 线程池中的线程要从任务队列中拿任务，所以任务队列中必须要先有任务，必须要加锁循环检测，如果任务队列为空，那么该线程应该进行等待，直到任务队列中有任务时再将其唤醒，这些操作都是通过加锁和条件变量完成的。
3. 主线程向任务队列中push一个任务后，此时可能有线程正处于等待状态，所以在新增任务后需要唤醒在条件变量下等待的线程。
4. 某线程从任务队列中拿到任务后，该任务就已经属于当前线程了，所以解锁之后再进行处理任务，让加锁的动作更细粒度，也因为处理任务的过程会耗费时间，所以不要将处理动作其放到临界区当中。
5. 要给执行线程函数用static修饰，这个函数的类型必须是void* (*callback)(void*)；如果放到类中，该函数就会多一个this指针。但是让他变成静态函数又不能访问线程池中的任务队列，所以要在线程创建的时候把线程池的对象指针传过去，因为初始化列表后已经有了对象，所以一定有this指针。也因为这个函数没有this指针，所以一些类内的操作要提供接口。

线程池的线程安全

        我们也可以把线程池变成单例模式（懒汉模式）的，让整个进程只有一个线程池，但是如果以后有多个线程同时访问，同时判断这个单例对象存不存在，那就会有线程安全的问题。

class ThreadPool
{
// ...
private: // 私有构造，删除拷贝构造和赋值重载ThreadPool(int thread_num = g_default_num) // 初始化后，就已经有了对象，也有了this指针:num_(thread_num){pthread_mutex_init(&lock_, nullptr);pthread_cond_init(&cond_, nullptr);for (int i = 0; i < num_; i++){threads_.push_back(new Thread(i + 1, routine, this) ); // 通过传入this指针就可以拿到ThreadPool中的task_queue}}ThreadPool(const ThreadPool<T>&) = delete;const ThreadPool<T> operator=(const ThreadPool<T>& ) = delete;public:static ThreadPool<T>* getThreadPool(int num = g_default_num) // 通过getThreadPool获取线程池{// 只有第一次为空的时候才创建，如果不为空直接返回thread_ptr，这样指针就只有一个{lockGuard lockguard(&mutex);if (nullptr == thread_ptr){thread_ptr = new ThreadPool<T>(num);}}return thread_ptr;}
// ...private:// 添加静态成员变量static ThreadPool<T>* thread_ptr; // 单例模式static pthread_mutex_t mutex;
};
// 初始化
template<class T>
ThreadPool<T>* ThreadPool<T>::thread_ptr = nullptr;template<class T>
pthread_mutex_t ThreadPool<T>::mutex = PTHREAD_MUTEX_INITIALIZER; 

        这样就可以保证，第一次获取ThreadPool对象的时候，多个线程访问就是安全的。但这就带来了另一个问题，如果每次想要获取ThreadPool对象的时候就会申请释放锁，这个行为也是在浪费资源，所以还要再调整一下。

static ThreadPool<T>* getThreadPool(int num = g_default_num)
{// 只有第一次为空的时候才创建，如果不为空直接返回thread_ptr，这样就只new了一次if (nullptr == thread_ptr) // 多判断一次不就可以了吗，已经创建了就直接返回，没有就加锁创建{lockGuard lockguard(&mutex);if (nullptr == thread_ptr){thread_ptr = new ThreadPool<T>(num);}}return thread_ptr;
}

这样使用双重判定空指针就减少了大量已经创建好单例，其他线程还在请求锁的行为。

日志文件的实现

我们需要用到下面这些接口。

// log.hpp
#pragma once#include <iostream>
#include <string>
#include <cstdio>
#include <cstdarg>
#include <unistd.h>
#include <sys/types.h>
#include <fcntl.h>
#include <ctime>// 日志级别
#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4const char* gLevelMap[] = {"DEBUG","NORMAL","WARNING","ERROR","FATAL"
};// 完整的日志功能，至少有：日志等级 时间 日志内容 支持用户自定义
void logMessage(int level, const char* format, ...) // 最后一个参数就是可变参数列表
{char stdBuffer[1024]; // 日志的标准部分time_t timestamp = time(nullptr); // 时间戳snprintf(stdBuffer, sizeof(stdBuffer), "[%s][%ld]", gLevelMap[level], timestamp);char logBuffer[1024]; // 自定义部分va_list args; // 可变参数列表va_start(args, format);vsnprintf(logBuffer, sizeof (logBuffer), format, args); // 用起来和printf相差不多va_end(args);// printf("%s%s\n", stdBuffer, logBuffer); // 打印到显示器FILE* fp = fopen("log.txt", "a");fprintf(fp, "%s%s\n", stdBuffer, logBuffer); // 打印到文件fclose(fp);
}

        所以以后如果要用到这些线程池、日志文件等，就直接用了。