# Ca1mThreadPool

**Repository Path**: ccjabc/ca1m-thread-pool

## Basic Information

- **Project Name**: Ca1mThreadPool
- **Description**: 线程池。（施磊老师课程）
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 1
- **Created**: 2025-10-21
- **Last Updated**: 2025-10-21

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

## C++11语言层级的线程池
- 基于C++11语言层的线程池
- 支持fixed和cached模式的线程池定制
- poolV1是手动实现Any类和Result类支持用户传递任意参数的任务函数（Visual Studio编译时需要设置为C++17标准）
- poolV2利用可变参模板编程和引用折叠原理，实现线程池addTask接口 — 支持用户传递任意参数的任务函数（C++14标准）
- PixelThreadPool是对poolV2进行CMake构建 - 支持用户传递任意参数的任务函数（C++14标准、Linux系统）

注：Windows平台下最好使用Visual Studio编译（MSVC），mingw64编译器线程模式不易修改使用（<mark>Thread model为posix时，支持默认的多线程使用；Thread model为win32时，支持的是Win32线程的API！</mark>）

### 基本概念
- 并发：单核，多个线程分配不同时间片 → “看起来共同执行”
- 并行：多核，真正的共同执行
- IO密集型：程序里面的指令涉及一些IO操作，比如设备、文件、网络操作（等待客户端连接IO操作是可以阻塞程序）
- CPU密集型：程序里面的指令都是做计算用的
- fixed模式线程池：线程池里面的线程个数是固定不变的，一般是ThreadPool创建时根据当前机器的CPU核心数量进行指定
- cached模式线程池：线程池里面的线程个数是可动态增长的，根据任务的数量动态的增加线程的数量

注：多线程程序一定好吗？不一定（<mark>对于多核系统而言，多线程程序对于IO密集型和CPU密集型都是适合的；对于单核系统而言，IO密集型程序是适合的，CPU密集型的程序不适合 - 线程的调度有额外开销。</mark>）

注：线程越多越好吗？不一定（线程的创建和销毁都是非常“重”的操作，业务端频繁创建和销毁线程资源消耗较大；线程栈本身占用大量内存，每个线程都需要线程栈；线程的上下文切换要占用大量时间。）

![image](./res/img1.png)

### threadpool.h
```cpp
#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <queue>
#include <memory>

// 任务抽象基类
class Task
{
public:
	// 用户可以自定义任意任务类型，从Task继承，重写run方法，实现自定义任务处理
	virtual void run() = 0;
};

// 线程池类型
class ThreadPool
{
public:
	// ...

private:
	// ...
	// std::vector<Thread*> threads_; // 智能指针 -> 自动析构裸指针
	std::vector<std::unique_ptr<Thread>> threads_; // 线程列表
	std::queue<std::shared_ptr<Task>> taskQue_; // 任务队列
};

#endif
```
- 定义任务抽象基类 → 方便用户自定义任务类型，实现自定义任务处理！
- 为什么线程列表用智能指针 → 自动析构裸指针！
- 为什么定义任务队列时不用`std::queue<Task*> taskQue_` → <mark>若用`Task*`指针，用户可能创建临时任务对象，提交任务函数`submitTask(concreteTask)`执行完成，任务对象析构，那线程池的基类指针获取已析构的对象，无法访问</mark> → **智能指针保证任务的生命周期！**

### threadpool.cpp
```cpp
#include "threadpool.h"

#include <iostream>
#include <thread>
#include <functional>

// 开启线程池
void ThreadPool::start(int initThreadSize)
{
	// 记录初始线程个数
	initThreadSize_ = initThreadSize;
	
	// 创建线程对象
	for (int i = 0; i < initThreadSize_; i++)
	{
		// 创建thread线程对象时，通过绑定器把线程函数给到thread线程对象
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this)); // 智能指针
		threads_.emplace_back(std::move(ptr));
	}

	// 启动所有线程 → 需要执行线程函数
	for (int i = 0; i < initThreadSize_; i++)
	{
		threads_[i]->start(); // 需要执行一个线程函数
	}
}

void ThreadPool::threadFunc()
{
	// ...
}

// 启动线程
void Thread::start()
{
	// 创建一个线程来执行一个线程函数
	std::thread t(func_); // C++11 线程对象t 和线程函数func_
	t.detach(); // 设置分离线程 → 防止线程对象出作用域自动析构
}
```
- 创建线程对象时，通过绑定器把线程函数和thread线程对象绑定 → <mark>将线程函数定义于ThreadPool类中，方便访问私有成员！</mark>

```cpp
// 给线程池提交任务 -> 用户调用该接口，传入任务对象，生产任务
void ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	// 获取锁
	std::unique_lock<std::mutex>lock(taskQueMtx_);

	// 线程通信 -> 等待任务队列有空余 wait wait_for 和 wait_until
	// while (taskQue_.size() == taskQueMaxThresHold_) { notFull_.wait(lock); } // 等价于底下
	// notFull_.wait(lock, [&]()->bool { return taskQue_.size() < taskQueMaxThresHold_; }); // 等待状态

	// 用户提交任务，最长不能阻塞超过1s，否则判断提交任务失败，返回
	if (!notFull_.wait_for(lock, std::chrono::seconds(1), [&]()->bool { return taskQue_.size() < taskQueMaxThresHold_; }))
	{
		// wait_for返回为false，说明用户提交任务时最长阻塞超过了1s
		std::cerr << "task queue is full, submit task fail." << std::endl;
		return;
	}


	// 若有空余，将任务放于任务队列中
	taskQue_.emplace(sp);
	taskSize_++;

	// 新放任务后，即可在notEmpty_上进行通知 -> 方便“消费者”消费
	notEmpty_.notify_all();
}
```
- wait -> 条件变量一直在等
- wait_for -> 条件变量等待时限制时间
- wait_until -> 条件变量等到某个时间
- 此处使用wait_for达到用户提交最长阻塞不超过1s的目的，且wait_for有bool类型返回值

### 构建可以接收任意类型数据的Any类
- C++17提供了Any类
- 如何让一个类型指向其它任意的类型 -> <mark>利用基类和派生类（数据包含在派生类中），结合模板</mark>
- Any类中定义Base*基类指针，基类指针可以指向其派生类对象
```cpp
// Any类型：可以接收任意类型的数据 -> 牛逼！
class Any
{
public:
	Any() = default;
	~Any() = default;
	Any(const Any&) = delete;
	Any& operator=(const Any&) = delete;
	Any(Any&&) = default;
	Any& operator = (Any&&) = default;

	// 此构造函数可以让Any类型接收任意其它的数据
	template<typename T>
	Any(T data) : base_(std::make_unique<Derive<T>>(data))
	{}

	// 获取Any对象里面存储的数据
	template<typename T>
	T cast_()
	{
		// 获取base_所指向派生类Derive对象的data成员变量
		Derive<T>* pd = dynamic_cast<Derive<T>*>(base.get()); // 基类指针转为派生类指针 RTTI
		if (pd == nullptr)
		{
			throw "type is unmatch!";
		}
		return pd->data_;
	}

private:
	// 基类类型
	class Base
	{
	public:
		virtual ~Base() = default;
	};

	// 派生类类型
	template<typename T>
	class Derive : public Base
	{
	public:
		Derive(T data) : data_(data)
		{}

		T data_; // 保存任意的其它类型
	};
private:
	// 定义一个基类的指针
	std::unique_ptr<Base> base;
};
```

注：Any类在线程池中的应用 → <mark>用户需要执行计算任务，但返回值类型不一定固定</mark>，而线程池中run函数是虚函数，无法和模板一起使用，故需设计Any类！

### 基于可变参模板编程和future类优化
- 使用future来代替Result -> 节省线程池代码（<mark>future类型定制用户提交任务的返回值</mark>）
- 基于可变参模板支持用户提交各种不同类型的任务！