【linux】手搓线程池
【linux】进程池
- 线程池:
一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着 监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利 用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
线程池的应用场景：
- 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技 术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个 Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情 况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限， 出现错误.
#ifndef __THREAD_HPP__
#define __THREAD_HPP__
#include <iostream>
#include <pthread.h>
#include <string>
#include <unistd.h>
#include <functional>
namespace ThreadModule
{
using func_t = std::function<void(std::string)>;
class Thread
{
public:
Thread(func_t func, std::string threadname = "none_name") // 为什么--常量引用和非常量引用的概念
: _func(func), _threadname(threadname), _stop(true)
{}
void Excute()
{
_func(_threadname);
}
static void *handler(void *args)
{
Thread *self = static_cast<Thread*>(args);
self->Excute();
return nullptr;
}
bool Start()
{
int ret = pthread_create(&_tid, nullptr, handler, this);
if (ret == 0)
{
_stop = false;
return true;
}
else
{
return false;
}
}
void Join()
{
if (!_stop)
{
pthread_join(_tid, nullptr);
}
}
std::string name()
{
return _threadname;
}
~Thread() {}
private:
std::string _threadname;
pthread_t _tid;
func_t _func;
bool _stop;
};
}
#endif
#ifndef __THREAD_POOL_HPP__
#define __THREAD_POOL_HPP__
#include <iostream>
#include <pthread.h>
#include <string>
#include <unistd.h>
#include <vector>
#include "log.hpp"
#include <queue>
#include "Thread.hpp"
#include <functional>
using namespace ThreadModule;
const int def_thread_sum = 3; // 默认线程数量
template <typename T>
class thread_pool
{
public:
void Lock_queue()
{
pthread_mutex_lock(&_mutex);
}
void UnLock_queue()
{
pthread_mutex_unlock(&_mutex);
}
void Wait_thread()
{
pthread_cond_wait(&_cond, &_mutex);
}
void Wake_up_thread()
{
pthread_cond_signal(&_cond);
}
void Wake_up_allthread()
{
pthread_cond_broadcast(&_cond);
}
public:
thread_pool(int threadsum = def_thread_sum)
: _threadsum(threadsum), _isruning(false), _waitsum(0)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
}
void HandlerTask(std::string name)// 类的成员方法，也可以成为另一个类的回调方法，方便我们继续类级别的互相调用！
{
while (true)
{
// 1. 保证队列安全
Lock_queue();
// 当任务队列为空，线程运行时
// 2. 队列中不一定有数据
while (_Task_queue.empty() && _isruning) // 当任务队列为空，有线程在运行，每个被唤醒的线程都要重新判断条件
{
_waitsum++; // 等待的线程数量加一
Wait_thread(); // 将线程加入等待队列中，解锁--等待被唤醒--加锁
_waitsum--;
}
// 2.1 如果线程池已经退出了 && 任务队列是空的
if (_Task_queue.empty() && !_isruning)
{
UnLock_queue();
break;
}
// 2.2 如果线程池不退出 && 任务队列不是空的
// 2.3 如果线程池已经退出 && 任务队列不是空的 --- 处理完所有的任务，然后在退出
// 3. 一定有任务, 处理任务
// 一定有任务
T t = _Task_queue.front(); // 取出任务
_Task_queue.pop();
UnLock_queue();
LOG(DEBUG, "%s get a task", name.c_str());//打印日志
// 4. 处理任务，这个任务属于线程独占的任务
t();
// 执行任务
LOG(DEBUG, "%s handler a task, result is: %s", name.c_str(), t.Result_tostring().c_str());
}
}
void Init_thread_pool()
{
// 指向构建出所有的线程，并不启动
for (int i = 0; i < _threadsum; i++)
{
std::string name = "thread-" + std::to_string(i + 1);
_thread_pool.emplace_back(std::bind(&thread_pool::HandlerTask, this, std::placeholders::_1), name);
LOG(INFO, "init thread %s done", name.c_str());//打印日志
}
_isruning = true;
}
void Start_threads()
{
for (auto &thread : _thread_pool)
{
thread.Start();
}
}
bool Enqueue(T &in)
{
bool ret = false;
if (_isruning)
{
Lock_queue();
_Task_queue.push(in);
if (_waitsum > 0) //等待队列中的线程数量大于0
{
Wake_up_thread();//唤醒等待队里中的一个线程
}
LOG(DEBUG, "enqueue task success");//打印日志
ret = true;
UnLock_queue();
}
return ret;
}
void WaitAllThread()
{
for (auto &thread : _thread_pool)
{
thread.Join();//等待所有线程
LOG(INFO, "%s is quit...", thread.name().c_str());//打印日志
}
}
void Stop()
{
Lock_queue();
_isruning = false;
Wake_up_allthread();
UnLock_queue();
}
~thread_pool()
{
pthread_mutex_destroy(&_mutex);//释放互斥量
pthread_cond_destroy(&_cond);//释放条件变量
}
private:
int _threadsum; //线程数量
std::vector<Thread> _thread_pool;//线程池结构
std::queue<T> _Task_queue;//任务队列
pthread_mutex_t _mutex;//互斥量
pthread_cond_t _cond;//条件变量
bool _isruning;//正在运行的队列
int _waitsum;//正在等待的线程的数量
};
#endif
#pragma once
#include <iostream>
#include<string>
class Task
{
public:
Task(int a, int b) : _a(a), _b(b)
{
}
void Excute()
{
_result = _a + _b;
}
void operator()()
{
_result = _a + _b;
}
std::string Result_tostring()
{
return std::to_string(_a)+" + " +std::to_string(_b)+" = "+std::to_string(_result);
}
std::string Debug_tostring()
{
return std::to_string(_a)+" + "+std::to_string(_b)+" = ?";
}
~Task()
{}
private:
int _a;
int _b;
int _result;
};
#pragma once
#include <iostream>
#include <stdarg.h>
#include <fstream>
#include "LockGuard.hpp"
const static char *logname = "log.txt";//日志文件
bool g_save = false;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
enum level//日志的等级
{
DEBUG = 0,
INFO,
WARNING,
ERROR,
FATAL
};
void save_file(const std::string &logname, std::string &massage)//保存日志到文件中
{
std::ofstream infile("logname", std::ios::app);
if (!infile.is_open())
{
return;
}
infile << massage << std::endl;
infile.close();
}
// 获取日志等级
std::string get_level_string(int level)
{
switch (level)
{
case DEBUG:
return "DEBUG";
case INFO:
return "INFO";
case WARNING:
return "WARNING";
case ERROR:
return "ERROR";
case FATAL:
return "FATAL";
default:
return "None";
}
}
// 获取时间字符串
std::string get_time_string()
{
time_t cur_time = time(nullptr);
if (cur_time == (time_t)-1)
{
printf("Failed to get the current time.\n");
return "None";
}
struct tm *formate_time = localtime(&cur_time);
if (formate_time == nullptr)
{
return "None";
}
char buffer[1024];
snprintf(buffer, sizeof(buffer), "%d-%d-%d %d:%d:%d",
formate_time->tm_year + 1900,
formate_time->tm_mon + 1,
formate_time->tm_mday,
formate_time->tm_hour,
formate_time->tm_min,
formate_time->tm_sec);
return buffer;
}
// 日志信息
void Log_inf(std::string filename, int line, bool is_save, int level, const char *format, ...)
{
std::string levelstr = get_level_string(level);
std::string time = get_time_string();
pid_t selfid = getpid();
char buffer[1024];
va_list arg;
va_start(arg, format);
vsnprintf(buffer, sizeof(buffer), format, arg);
va_end(arg);
std::string massage = "[" + time + "]" + "[" + levelstr + "]" + "[" + std::to_string(selfid) + "]" + "[" + filename + "]" + "[" + std::to_string(line) + "]" + buffer + "\n";
LockGuard lockguard(mutex); // RAII
if (is_save)
{
// 保存到文件中
save_file(logname, massage);
}
else
{ //向屏幕中打印
std::cout << massage;
}
}
// 定义宏
#define LOG(level, format, ...) \
do \
{ \
Log_inf(__FILE__, __LINE__, g_save, level, format, ##__VA_ARGS__); \
} while (0)
#define Enablefile() \
do \
{ \
g_save = true; \
} while (0)
#define EnableScrean() \
do \
{ \
g_save = false; \
} while (0)
#include <iostream>
#include "ThreadPool.hpp"
#include "Thread.hpp"
#include"Task.hpp"
#include<memory>
#include"log.hpp"
using namespace ThreadModule;
int main()
{
Enablefile();
std::unique_ptr<thread_pool<Task>> tp = std::make_unique<thread_pool<Task>>(5);
tp->Init_thread_pool();
tp->Start_threads();
int cnt = 10;
srand((signed)time(NULL));
while(cnt--)
{
int frist = rand()%10+1;
usleep(1234);
int second = rand()%10+1;
Task t(frist,second);
LOG(INFO,"main send Task is :%s", t.Debug_tostring().c_str());//打印日志
tp->Enqueue(t);
sleep(2);
}
tp->Stop();
tp->WaitAllThread();
LOG(INFO,"main thread is quit ... \n");//打印日志
return 0;
}
[!IMPORTANT]
【Linux】线程篇完结！！，下一篇【Linux】网络篇