目录
一、线程概念
1.linux线程如何理解？
linux下一个进程是由一个PCB块task_struct，进程地址空间mm_struct，页表+MMU，代码和数据组成，而创建子进程，因为进程具有独立性，子进程也有自己的task_struct,mm_struct,页表。而我们创建出一个只有一个PCB的东西，它指向父进程的进程地址空间，可以通过页表访问父进程的代码和数据，我们把这个称作线程。
linux实现方案
- 在linux中，线程在进程"内部"执行，线程在进程的程序地址空间内运行。(任何执行流要执行，都要有资源，进程地址空间就是进程的资源窗口)
- 在linux中，线程的执行粒度要比进程更细呢？线程执行进程代码的一部分。
2.重新定义线程、进程
1.什么是线程？
- 线程时操作系统调度的基本单位
- 线程是进程内部的执行流资源
2.什么是进程？
- 内核观点：进程是承担分配系统资源的基本实体
线程是进程内部的执行流资源
3.linux下线程是如何设计的？
根据linux一贯的设计哲学 --- 先描述，再组织。先看windows下是怎么设计，windows下给线程单独设计了一个 struct tcb;//线程控制块。这会导致代码，进程与线程关系极其复杂，难以捋清楚。而linux下采用的方法是复用task_struct模拟线程。linux下没有真正意义上的线程，而是用"进程内核数据结构"模拟的线程。
CPU不关心是进程还是线程，但是它看到的执行流 <= 进程。
linux中的执行流，是轻量级进程。
4.重谈地址空间
1.虚拟地址是如何转换为物理地址的呢？
假设虚拟地址是32位的。
这个虚拟地址通常会被划分为3部分，32 = 10 + 10 + 12；
- 第一个10位，[0, 2^10-1]个地址范围，会来索引页目录，前10位会成为页目录的下标，页目录一共有1024个页目录表项，页目录表项中存放二级页表的首地址。
- 中间的10位，会成为二级页表的下标，二级页表一共有1024个页表表项，每个页表表项存放物理内存页框的起始地址。
- 最后12位，[0，2^12 -1]个地址范围，会成为页框的偏移量，每个页框大小刚好是4kb = 2^12，而所要查找的内容起始地址 = 当前页框起始地址 + 页框的偏移量。
注：二级页表大部分情况下都是不全的
但是，即使如此简化页表，创建一个进程依旧是一个很繁重的工作。
2.如何理解资源分配？
线程分配资源，本质是划分地址空间范围
5. linux线程周边概念
1.线程要比进程更轻量化吗
整个生命周期
- 创建和释放更轻量化（生死）
- 切换更轻量化（运行）
- 线程内的切换，不需要重新缓存cache数据（硬件）
PCB内有身份标识，最开始创建的PCB是主线程，其它的是新线程
2.线程的优点
- 创建一个新线程的代价要比创建一个新进程小得多
- 与进程之间的切换相比，线程之间的切换需要操作系统做的工作要少很多
- 线程占用的资源要比进程少很多
- 能充分利用多处理器的可并行数量
- 在等待慢速I/O操作结束的同时，程序可执行其他的计算任务
- 计算密集型应用，为了能在多处理器系统上运行，将计算分解到多个线程中实现
- I/O密集型应用，为了提高性能，将I/O操作重叠。线程可以同时等待不同的I/O操作。
3.线程的缺点
- 性能损失
- 一个很少被外部事件阻塞的计算密集型线程往往无法与共它线程共享同一个处理器。如果计算密集型 线程的数量比可用的处理器多，那么可能会有较大的性能损失，这里的性能损失指的是增加了额外的同步和调度开销，而可用的资源不变。
- 健壮性降低
- 编写多线程需要更全面更深入的考虑，在一个多线程程序里，因时间分配上的细微偏差或者因共享了 不该共享的变量而造成不良影响的可能性是很大的，换句话说线程之间是缺乏保护的。
- 缺乏访问控制
- 进程是访问控制的基本粒度，在一个线程中调用某些OS函数会对整个进程造成影响。
- 编程难度提高
- 编写与调试一个多线程程序比单线程程序困难得多
4.线程异常
- 单个线程如果出现除零，野指针问题导致线程崩溃，进程也会随着崩溃
- 线程是进程的执行分支，线程出异常，就类似进程出异常，进而触发信号机制，终止进程，进程终止，该进程内的所有线程也就随即退出
5.线程用途
- 合理的使用多线程，能提高CPU密集型程序的执行效率
- 合理的使用多线程，能提高IO密集型程序的用户体验（如生活中我们一边写代码一边下载开发工具，就是多线程运行的一种表现）
6.进程vs线程
- 进程是资源分配的基本单位
- 线程是调度的基本单位
- 线程共享进程数据，但也拥有自己的一部分数据:
- 线程ID
- 一组寄存器 //线程的上下文
- 栈 //线程需要有独立的栈结构
- errno
- 信号屏蔽字
- 调度优先级
进程的多个线程共享 同一地址空间,因此Text Segment、Data Segment都是共享的,如果定义一个函数,在各线程中都可以调用,如果定义一个全局变量,在各线程中都可以访问到,除此之外,各线程还共享以下进程资源和环境:
- 文件描述符表 //重要
- 每种信号的处理方式(SIG_ IGN、SIG_ DFL或者自定义的信号处理函数)
- 当前工作目录
- 用户id和组id
二、线程控制
1.线程的创建
#include <pthread.h> int pthread_create(pthread_t *thread //输出型参数 , const pthread_attr_t *attr //创建线程的属性，大部分时间设为nullptr即可 , void *(*start_routine) (void *) //函数指针 , void *arg); //创建线程成功，新线程回调线程函数的时候，需要参数，这个参数就是给 线程函数传递的 Compile and link with -pthread.示例代码
void* threadRoutine(void* args) { while(true) { cout << "new thread, pid: " << getpid() << endl; sleep(1); } } int main() { pthread_t tid; pthread_create(&tid, nullptr, threadRoutine, nullptr); while(true) { cout << "main thread, pid: " << getpid() << endl; sleep(1); } return 0; }
2.线程的等待
#include <pthread.h> int pthread_join(pthread_t thread, void **retval);示例代码
void* threadRoutine(void* args) { const char *name = (const char*)args; int cnt = 5; while (true) { //cout << "%s, pid: " << getpid() << endl; printf("%s, pid: %d\n", name, getpid()); sleep(1); cnt--; if(cnt == 0) break; } // exit(1); //exit是用来终止进程的，不能用来终止线程 return (void *)100; // 走到这里，默认线程退出了 } int main() { pthread_t tid; pthread_create(&tid, nullptr, threadRoutine, (void*)"Thread 1"); //不是系统调用 sleep(7); void *ret; pthread_join(tid, &ret); // main thread等待时，是默认阻塞等待的 cout << "main thread quit...ret: " << (long long int)ret << endl; return 0; }
3.线程的终止
#include <pthread.h> void pthread_exit(void *retval);线程终止方式有3个
- pthread_exit
- pthread_cancel
- return
注：exit是用来终止进程的，不能用来终止线程
4.线程的取消
#include <pthread.h>
int pthread_cancel(pthread_t thread);
如果线程被取消，那么它会返回一个PTHREAD_ CANCELED，这是个宏，值为-1
5.获取线程的id
#include <pthread.h>
pthread_t pthread_self(void);
6.线程相关函数参数细节再深入
线程的参数和返回值，不仅仅可以传递一般参数，也可以传递对象
struct Requst { int _start; int _end; std::string _threadname; Requst(int start, int end, const std::string& threadname) : _start(start) , _end(end) , _threadname(threadname) {} int Run() { int num = 0; for (size_t i = _start; i <= _end; ++i) { cout << _threadname << "is running, i: " << i << endl; num += i; usleep(10000); } return num; } }; struct Response { int _result; int _exitcode; Response(int result, int exitcode) : _result(result) , _exitcode(exitcode) {} }; void *sumCount(void* args) //线程的参数和返回值，不仅仅可以传递一般参数，也可以传递对象 { Requst *rq = static_cast<Requst *>(args); Response *rsp = new Response(0, 0); rsp->_result = rq->Run(); delete rq; return rsp; } int main() { pthread_t tid; Requst *rq = new Requst(1, 100, "thread 1"); pthread_create(&tid, nullptr, sumCount, rq); void *ret; pthread_join(tid, &ret); Response *rsp = static_cast<Response *>(ret); cout << "rsp->_reslut: " << rsp->_result << " rsp->_exitcode: " << rsp->_exitcode << endl; return 0; }
zz
7.线程库
目前，我们学习到的线程都是linux下的pthread库里，该库叫做原生线程库
而c++11已经引入线程库了，只需
#include <thread>c++语言具有跨平台性，在linux系统下，c++线程库底层就是使用pthread进行封装的。
clone函数，与fork底层结构类似，用来生成轻量化进程的函数
#define _GNU_SOURCE #include <sched.h> int clone(int (*fn)(void *), void *child_stack, int flags, void *arg, ... /* pid_t *ptid, void *newtls, pid_t *ctid */ );线程的概念是库给我们1维护的 -> 线程库为维护线程概念 -- 不用维护线程的执行流
我们使用的原生线程库，都是加载到内存当中！
内核中没有很明确的线程的概念。
轻量化进程的概念：
系统不会给我们直接提供线程的系统调用，只会给我们提供轻量级进程的系统调用！
每一个线程的库级别的tcb的起始地址，叫做线程的tid！
除了主线程，所有其它线程的独立栈，都在共享区。具体来讲都在pthread.so动态库中，tid指向用户的tcb中
每个线程都有自己的独立栈，线程与线程之间几乎没有秘密，线程的栈上的数据也是能被其它线程看到并且可以访问的。
怎么定义一个私有的全局变量？
当我们定义一个全局变量，在定义变量前加上__thread，这是一个编译选项，就会给每个线程专门开辟一份，每一个线程中使用这个变量时，它的地址不相同，这叫做线程的局部存储
而这个__thread只能定义内置类型，不能定义自定义类型
8.分离线程
- 默认情况下，新创建的线程是joinable的，线程退出后，需要对其进行pthread_join操作，否则无法释放资源，从而造成系统泄漏。
- 如果不关心线程的返回值，join是一种负担，这个时候，我们可以告诉系统，当线程退出时，自动释放线程资源。
#include <pthread.h>
int pthread_detach(pthread_t thread);
线程可以由主线程进行分离，也可以由其它线程主动分离 -- pthread_detach(pthread_self());
注：joinable和分离是冲突的，一个线程不能既是joinable又是分离的。
三、linux线程互斥和同步
来看一段模拟抢票的代码
#define NUM 3 int ticket = 3; struct thread_data { string threadname; thread_data(int number) { threadname = "thread-" + std::to_string(number); } }; void *getTicket(void *args) { thread_data *td = (thread_data *)args; const char *name = td->threadname.c_str(); while(true) { if(ticket > 0) { usleep(1000); cout << name << " get a ticket : " << ticket << endl; --ticket; } else break; } printf("%s quit ...\n", name); return nullptr; } int main() { vector<pthread_t> tids; vector<thread_data*> thread_datas; for (size_t i = 1; i <= NUM; ++i) { pthread_t tid; thread_data *td = new thread_data(i); thread_datas.push_back(td); pthread_create(&tid, nullptr, getTicket, thread_datas[i - 1]); tids.push_back(tid); } for(auto tid : tids) { pthread_join(tid, nullptr); } for (auto td : thread_datas) { delete td; } return 0; }
结果竟然出现了负数，一共3张票，被卖了4张
对一个全局变量，进行多线程访问，进行修改，是否是安全的吗？答案是不安全
一般对这种数据进行计算时，通常是3步
- 先将自己代码段中ticket读入到CPU内部的寄存器当中
- CPU内部进行操作
- 将计算结果写回内存
寄存器不等于寄存器的内容
线程在执行时，将共享数据加载到寄存器的本质是：把数据的内容，变成自己的上下文，以拷贝的方式，给自己单独拿了一份。
当一个线程将加载到寄存器的数据进行操作后，他会把数据拿走，变成自己的上下文。而另一个线程来临时，他会把恢复自己的上下文，导致数据不一致问题。
而多个线程同时进入执行ticket--这步操作时
1.互斥
对共享数据的任何访问，保证任何时候都只有一个数据流来访问 --- 互斥 -- 锁 ！
1.初始化锁和销毁锁
#include <pthread.h>
int pthread_mutex_destroy(pthread_mutex_t *mutex);
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; //全局初始化
2.加锁和解锁
#include <pthread.h>
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex); //非阻塞加锁
int pthread_mutex_unlock(pthread_mutex_t *mutex);
加锁的本质是：CPU内部有一个寄存器eax，值为1，这就是锁，而当线程想来申请锁时，先查看这个寄存器的值是否大于0，若大于0，则将这个寄存器的值与自己内存当中的一个值进行了交换，而这个值为0，否则挂起等待。
交换的本质是：把内存中的数据交换到CPU的寄存器中。
把一个共享的锁，让一个线程以一条汇编的方式，交换到自己的上下文当中，就说明当前线程持有锁了。
解锁
2.可重入vs线程安全
1.线程安全：
- 每个线程对全局变量或者静态变量只有读取的权限，而没有写入的权限，一般来说这些线程是安全的
- 类或者接口对于线程来说都是原子操作
- 多个线程之间的切换不会导致该接口的执行结果存在二义性
2.不可重入情况
- 调用了malloc/free函数，因为malloc函数是用全局链表来管理堆的
- 调用了标准I/O库函数，标准I/O库的很多实现都以不可重入的方式使用全局数据结构
- 可重入函数体内使用了静态的数据结构
3.常见可重入的情况
- 不使用全局变量或静态变量
- 不使用用malloc或者new开辟出的空间
- 不调用不可重入函数
- 不返回静态或全局数据，所有数据都有函数的调用者提供
- 使用本地数据，或者通过制作全局数据的本地拷贝来保护全局数据
4.可重入与线程安全联系
- 函数是可重入的，那就是线程安全的
- 函数是不可重入的，那就不能由多个线程使用，有可能引发线程安全问题
- 如果一个函数中有全局变量，那么这个函数既不是线程安全也不是可重入的。
5.可重入与线程安全区别
- 可重入函数是线程安全函数的一种
- 线程安全不一定是可重入的，而可重入函数则一定是线程安全的。
- 如果将对临界资源的访问加上锁，则这个函数是线程安全的，但如果这个重入函数若锁还未释放则会产生 死锁，因此是不可重入的。
3.同步
1.概念
对于上述抢票，即便是加锁后，让线程单独访问同一份资源，互不干扰。但是结果却是，一份进程抢票抢到爽，其它线程却拿不到锁，进而导致饥饿问题。这是我们给线程按队列访问同一份共享资源，这就叫做同步。
同步问题是保证数据安全的情况下，让我们的线程访问的资源具有一定的顺序性。
在上面这个临界区中，线程可以被切换吗？
在线程被切出去的时候，是持有锁被切走的。不在期间，没有人能进入临界区访问临界资源！！
对于其它线程来说，一个线程要么申请锁，要么释放锁
当前线程访问临界区的时候，对于其它线程来说是原子的
死锁
1.死锁产生的必要条件
- 互斥条件：一个资源每次只能被一个执行流使用
- 请求与保持条件：一个执行流因请求资源而阻塞时，对已获得的资源保持不放
- 不剥夺条件:一个执行流已获得的资源，在末使用完之前，不能强行剥夺
- 循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系
2.解决死锁的方法
理念：破坏上述四个必要条件，原则上只需一个不满足。
- 破坏死锁的四个必要条件
- 加锁顺序一致
- 避免锁未释放的场景
- 资源一次性分配
2. 条件变量
- 当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
- 例如一个线程访问队列时，发现队列为空，它只能等待，只到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。
1.初始化
int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr); 参数： cond：要初始化的条件变量 attr：NULL pthread_cond_t cond = PTHREAD_COND_INITIALIZER; //全局初始化2.销毁
int pthread_cond_destroy(pthread_cond_t *cond)3.等待条件满足
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex); 参数： cond：要在这个条件变量上等待 mutex：互斥量4.唤醒等待
int pthread_cond_broadcast(pthread_cond_t *cond); //唤醒所有 int pthread_cond_signal(pthread_cond_t *cond); //唤醒一个示例
int cnt = 0; pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; pthread_cond_t cond = PTHREAD_COND_INITIALIZER; void* Count(void *args) { pthread_detach(pthread_self()); uint64_t number = (uint64_t)args; cout << "pthread-" << number << " create success" << endl; while (true) { pthread_mutex_init(&mutex, nullptr); pthread_cond_wait(&cond, &mutex); //1.为什么在这里等待？ pthread_cond_wait在让线程等待时，会自动释放锁 cout << "pthread-" << number << " cnt: " << cnt++ << endl; pthread_mutex_destroy(&mutex); } } int main() { for (size_t i = 0; i < 5; ++i) { pthread_t tid; pthread_create(&tid, nullptr, Count, (void *)i); usleep(1000); } sleep(3); cout << "main thread ctrl begin ..." << endl; while(true) { sleep(1); pthread_cond_signal(&cond); // 唤醒在等待队列中等待的一个线程，默认都是第一个 cout << "signal one thread..." << endl; } while(true) sleep(1); return 0; }
3.竞争条件
竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理
四、cp问题
consumer producter ， 生产者 -- 消费者模型
为何要使用生产者消费者模型 生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产 者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理， 直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了 生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。
生产者消费者模型是一种多进程、多线程同步互斥的一种策略，一共有3种关系，分别是生产者和生产者的关系，消费者和消费者的关系，生产者和消费者的关系，有2种角色，一个是生产者，一个是消费者，1个交易场所，一种特殊结构的内存空间。
优点：
- 支持忙闲不均
- 生产和消费解耦
生产者消费者代码实现
template<class T>
class BlockQueue
{
static const int defaultcup = 20;
public:
BlockQueue(int maxcup = defaultcup)
: _maxcup(maxcup)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_c_cond, nullptr);
pthread_cond_init(&_p_cond, nullptr);
_low_water = _maxcup / 3;
_high_water = (_maxcup * 2) / 3;
}
~BlockQueue()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_c_cond);
pthread_cond_destroy(&_p_cond);
}
T pop()
{
pthread_mutex_lock(&_mutex);
if(_q.size() == 0)
{
pthread_cond_wait(&_c_cond, &_mutex);
}
T tmp = _q.front();
_q.pop();
if(_q.size() < _low_water)
pthread_cond_signal(&_p_cond);
pthread_mutex_unlock(&_mutex);
return tmp;
}
void push(const T& in)
{
pthread_mutex_lock(&_mutex);
if(_q.size() == _maxcup)
{
pthread_cond_wait(&_p_cond, &_mutex); //调用的时候，会自动释放锁
}
//1.队列没满 2.被唤醒
_q.push(in);
if(_q.size() > _high_water)
pthread_cond_signal(&_c_cond);
pthread_mutex_unlock(&_mutex);
}
private:
std::queue<T> _q; //共享资源
int _maxcup; //极值
pthread_mutex_t _mutex;
pthread_cond_t _c_cond;
pthread_cond_t _p_cond;
//策略 -- 调整生产和消费
int _low_water;
int _high_water;
};五、POSIX信号量
POSIX信号量和SystemV信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的。 但POSIX可以用 于线程间同步。
信号量的本质是一把计数器。计数器的本质是临界资源的数量。
1.初始化信号量
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
参数：
pshared:0表示线程间共享，非零表示进程间共享
value：信号量初始值
2.销毁信号量
int sem_destroy(sem_t *sem);
3.等待信号量 == P()
功能：等待信号量，会将信号量的值减1
int sem_wait(sem_t *sem);4.发布信号量 == V()
功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1。
int sem_post(sem_t *sem);六、基于环形的生产消费模型
- 环形队列采用数组模拟，用模运算来模拟环状特性
- 环形结构起始状态和结束状态都是一样的，不好判断为空或者为满，所以可以通过加计数器或者标记位来 判断满或者空。另外也可以预留一个空的位置，作为满的状态
- 但是我们现在有信号量这个计数器，就很简单的进行多线程间的同步过程
template<class T>
class RingQueue
{
const static int dafaultcup = 5;
void P(sem_t &sem)
{
sem_wait(&sem);
}
void V(sem_t &sem)
{
sem_post(&sem);
}
void Lock(pthread_mutex_t &mutex)
{
pthread_mutex_lock(&mutex);
}
void UnLock(pthread_mutex_t &mutex)
{
pthread_mutex_unlock(&mutex);
}
public:
RingQueue(int cup = dafaultcup)
: _ringqueue(cup)
, _cup(cup)
, _c_step(0)
, _p_step(0)
{
sem_init(&_cdata_sem, 0, 0);
sem_init(&_pspace_sem, 0, _cup);
pthread_mutex_init(&_c_mutex, nullptr);
pthread_mutex_init(&_p_mutex, nullptr);
}
~RingQueue()
{
sem_destroy(&_cdata_sem);
sem_destroy(&_pspace_sem);
pthread_mutex_destroy(&_c_mutex);
pthread_mutex_destroy(&_p_mutex);
}
void push(const T& in)
{
P(_pspace_sem);
Lock(_p_mutex);
_ringqueue[_p_step] = in;
//位置后移，维持环形队列
++_p_step;
_p_step %= _cup;
UnLock(_p_mutex);
V(_cdata_sem);
}
void pop(T* out)
{
P(_cdata_sem);
Lock(_c_mutex);
*out = _ringqueue[_c_step];
++_c_step;
_c_step %= _cup;
UnLock(_c_mutex);
V(_pspace_sem);
}
private:
vector<T> _ringqueue; //模拟环形队列
int _cup; //极值
int _c_step; //消费者下标
int _p_step; //生产者下标
sem_t _cdata_sem; //消费者关心的数据资源
sem_t _pspace_sem; //生产者关心的空间资源
//两把锁
pthread_mutex_t _c_mutex, _p_mutex;
};七、线程池
1.池化技术
理念是以空间换时间
2.c++类内创建线程
以后单独做一篇博客
3.线程池概念
一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量。
4.线程池的应用场景：
- 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB服务器完成网页请求这样的任务，使用线程池技术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个Telnet连接请求，线程池的优点就不明显了。因为Telnet会话时间比线程的创建时间大多了。
- 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
- 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，出现错误.
5.线程池的种类：
6.线程池示例：
- 创建固定数量线程池，循环从任务队列中获取任务对象，
- 获取到任务对象后，执行任务对象中的任务接口
7.代码示例
ThreadPool.hpp
#pragma once #include <iostream> #include <string> #include <vector> #include <queue> #include <pthread.h> #include <ctime> #include <unistd.h> #include "Task.hpp" struct thread { pthread_t _tid; std::string _name; }; static const int defaultnum = 5; template<class T> class ThreadPool { public: void Lock() { pthread_mutex_lock(&_mutex); } void UnLock() { pthread_mutex_unlock(&_mutex); } void WeakUp() { pthread_cond_signal(&_cond); } void ThreadSleep() { pthread_cond_wait(&_cond, &_mutex); } bool IsQueueEmpty() { return _q.empty(); } std::string GetThreadName(pthread_t tid) { for(const auto& ti : _threads) { if(ti._tid == tid) return ti._name; } return "None"; } public: ThreadPool(int num = defaultnum) : _threads(num) { pthread_mutex_init(&_mutex, nullptr); pthread_cond_init(&_cond, nullptr); } ~ThreadPool() { pthread_mutex_destroy(&_mutex); pthread_cond_destroy(&_cond); } //不允许有this指针，所以以static修饰 static void *HeadlerTask(void *args) { ThreadPool<Task> *tp = static_cast<ThreadPool<Task> *>(args); std::string name = tp->GetThreadName(pthread_self()); while(true) { tp->Lock(); while(tp->IsQueueEmpty()) { tp->ThreadSleep(); } T t = tp->Pop(); tp->UnLock(); t(); std::cout << name << " run, result: " << t.GetResult() << std::endl; } // 处理任务 } void Start() { int num = _threads.size(); for (size_t i = 0; i < num; ++i) { _threads[i]._name = "Thread-" + std::to_string(i + 1); pthread_create(&(_threads[i]._tid), nullptr, HeadlerTask, this); } } void Push(const T& in) { Lock(); _q.push(in); WeakUp(); UnLock(); } T Pop() { T t = _q.front(); _q.pop(); return t; } private: std::vector<thread> _threads; std::queue<Task> _q; pthread_mutex_t _mutex; pthread_cond_t _cond; };
main.cc
#include "ThreadPool.hpp" #include "Task.hpp" int main() { ThreadPool<Task> *tp = new ThreadPool<Task>(5); srand(time(nullptr) ^ getpid()); tp->Start(); while(true) { //1.构建任务 int x = rand() % 10 + 1; usleep(10); int y = rand() % 9; char op = opers[rand() % opers.size()]; //2.交给线程池处理 Task t(x, y, op); tp->Push(t); std::cout << "Main thread make a task: " << t.GetTask() << std::endl; sleep(1); } return 0; }
Task.hpp
#pragma once #include <iostream> #include <string> std::string opers="+-*/%"; enum{ DivZero=1, ModZero, Unknown }; class Task { public: Task() {} Task(int x, int y, char op) : data1_(x), data2_(y), oper_(op), result_(0), exitcode_(0) { } void run() { switch (oper_) { case '+': result_ = data1_ + data2_; break; case '-': result_ = data1_ - data2_; break; case '*': result_ = data1_ * data2_; break; case '/': { if(data2_ == 0) exitcode_ = DivZero; else result_ = data1_ / data2_; } break; case '%': { if(data2_ == 0) exitcode_ = ModZero; else result_ = data1_ % data2_; } break; default: exitcode_ = Unknown; break; } } void operator ()() { run(); } std::string GetResult() { std::string r = std::to_string(data1_); r += oper_; r += std::to_string(data2_); r += "="; r += std::to_string(result_); r += "[code: "; r += std::to_string(exitcode_); r += "]"; return r; } std::string GetTask() { std::string r = std::to_string(data1_); r += oper_; r += std::to_string(data2_); r += "=?"; return r; } ~Task() { } private: int data1_; int data2_; char oper_; int result_; int exitcode_; };
八、线程安全的单例模式
单例模式可以去我以往的博客C++特殊类的设计中查看。
将前面的线程池改为饿汉模式
struct thread
{
pthread_t _tid;
std::string _name;
};
static const int defaultnum = 5;
template<class T>
class ThreadPool
{
public:
void Lock()
{
pthread_mutex_lock(&_mutex);
}
void UnLock()
{
pthread_mutex_unlock(&_mutex);
}
void WeakUp()
{
pthread_cond_signal(&_cond);
}
void ThreadSleep()
{
pthread_cond_wait(&_cond, &_mutex);
}
bool IsQueueEmpty()
{
return _q.empty();
}
std::string GetThreadName(pthread_t tid)
{
for(const auto& ti : _threads)
{
if(ti._tid == tid)
return ti._name;
}
return "None";
}
public:
~ThreadPool()
{
pthread_mutex_destroy(&_mutex);
pthread_cond_destroy(&_cond);
}
//不允许有this指针，所以以static修饰
static void *HeadlerTask(void *args)
{
ThreadPool<Task> *tp = static_cast<ThreadPool<Task> *>(args);
std::string name = tp->GetThreadName(pthread_self());
while(true)
{
tp->Lock();
while(tp->IsQueueEmpty())
{
tp->ThreadSleep();
}
T t = tp->Pop();
tp->UnLock();
t();
std::cout << name << " run, result: " << t.GetResult() << std::endl;
}
// 处理任务
}
void Start()
{
int num = _threads.size();
for (size_t i = 0; i < num; ++i)
{
_threads[i]._name = "Thread-" + std::to_string(i + 1);
pthread_create(&(_threads[i]._tid), nullptr, HeadlerTask, this);
}
}
void Push(const T& in)
{
Lock();
_q.push(in);
WeakUp();
UnLock();
}
T Pop()
{
T t = _q.front();
_q.pop();
return t;
}
static ThreadPool<T>* GetIntance(int num = defaultnum)
{
if(nullptr == _tp)
{
pthread_mutex_lock(&_lock);
if (nullptr == _tp)
{
_tp = new ThreadPool<T>(num);
}
pthread_mutex_unlock(&_lock);
return _tp;
}
}
private:
ThreadPool(int num = defaultnum)
: _threads(num)
{
pthread_mutex_init(&_mutex, nullptr);
pthread_cond_init(&_cond, nullptr);
}
ThreadPool(const ThreadPool<T> &) = delete;
const ThreadPool<T> &operator=(const ThreadPool<T> &) = delete;
private:
std::vector<thread> _threads;
std::queue<Task> _q;
pthread_mutex_t _mutex;
pthread_cond_t _cond;
//单例模式 -- 饿汉模式
volatile static ThreadPool<T> *_tp;
static pthread_mutex_t _lock;
};
template <class T>
ThreadPool<T> *ThreadPool<T>::_tp = nullptr;
template <class T>
pthread_mutex_t ThreadPool<T>::_lock = PTHREAD_MUTEX_INITIALIZER;注意事项:
- 加锁解锁的位置
- 双重 if 判定, 避免不必要的锁竞争
- volatile关键字防止过度优化
九、STL,智能指针和线程安全
不是. 原因是, STL 的设计初衷是将性能挖掘到极致, 而一旦涉及到加锁保证线程安全, 会对性能造成巨大的影响. 而且对于不同的容器, 加锁方式的不同, 性能可能也不同(例如hash表的锁表和锁桶). 因此 STL 默认不是线程安全. 如果需要在多线程环境下使用, 往往需要调用者自行保证线程安全.