怎么选择昆明网站建设,全国信息企业公示系统官网,wordpress 主题失败,南阳理工网站建设文章目录 1. 线程概念1.1 概念1.2 理解#xff08;Linux OS角度#xff09;1.3 见一见 2. 线程优缺点3. 线程使用3.1 认识线程库3.2 使用3.2.1 线程创建3.2.2 线程等待3.2.3 线程退出3.2.4 线程取消3.2.5 获取线程id3.2.6 线程分离 3.3 理解线程库3.4 证明线程栈3.5 线程局部… 文章目录 1. 线程概念1.1 概念1.2 理解Linux OS角度1.3 见一见 2. 线程优缺点3. 线程使用3.1 认识线程库3.2 使用3.2.1 线程创建3.2.2 线程等待3.2.3 线程退出3.2.4 线程取消3.2.5 获取线程id3.2.6 线程分离 3.3 理解线程库3.4 证明线程栈3.5 线程局部存储 4. 线程互斥4.1 相关概念4.2 互斥场景4.3 代码演示模拟抢票4.4 细节4.5 锁的实现原理4.6 线程封装 5. 可重入函数和线程安全5.1 概念5.2 常见线程安全情况5.3 常见不可重入情况5.4 常见可重入情况5.5 可重入函数和线程安全关系5.6 可重入函数和线程安全的区别 6. 死锁6.1 四个充分必要条件6.2 避免死锁 7. 线程同步7.1 概念7.2 条件变量7.2.1 条件变量初始化7.2.2 条件变量销毁7.2.3 条件变量等待7.2.4 唤醒等待 7.3 demo 8. 生产者-消费者模型8.1 概念8.2 优点和关系8.3 BlockQueue生产者-消费者模型8.4 POSIX信号量8.4.1 信号量初始化8.4.2 信号量销毁8.4.3 等待信号量8.4.4 发布信号量 8.5 RingQueue生产者-消费者模型 9. 线程池threadpool_V1.hppthread_library.hppthreadpool_V2.hpplock_guard.hppthreadpool_V3.hpptask.hpptest.ccthreadpool_V4.hpptest_threadpoolV4.cc 10. 常见锁 1. 线程概念
1.1 概念
线程是一个执行分支执行粒度比进程更细调度成本更低线程是进程内部的一个执行流线程是CPU调度的基本单位进程是承担分配系统资源的基本实体
1.2 理解Linux OS角度 线程PCB是拷贝的进程PCB线程的PCB执行的是代码段中的某个函数区域也就相当于进程内部的一个执行流粒度也就更细谁来区分进程PCB和线程PCBOS。因为CPU硬件是给人打工的只有OS给CPU数据CPU才会去处理OS需不需要管理线程呢需要先描述再组织OS管理线程结构TCB线程控制块属于进程PCB。Linux中直接复用PCB结构体用PCB模拟线程TCB所以Linux中没有真正意义上的线程而是进程方案模拟的线程Linux中的线程叫做Linux轻量级进程
1.3 见一见
#include iostream
#include pthread.h
#include unistd.hvoid* thread1_run(void* args)
{while(true){printf(thread1: run....!\n);sleep(1);}
}void* thread2_run(void* args)
{while(true){printf(thread2: run....!\n);sleep(1);}
}void* thread3_run(void* args)
{while(true){printf(thread3: run....!\n);sleep(1);}
}int main()
{pthread_t thread1, thread2, thread3; pthread_create(thread1, NULL, thread1_run, NULL);pthread_create(thread1, NULL, thread2_run, NULL);pthread_create(thread1, NULL, thread3_run, NULL);while(true){printf(main thread: run....!\n);sleep(1);}return 0;
}//makefile
thread:thread.ccg -o $ $^ -stdc11 -lpthread
.PHONY:cleanrm -f thread运行现象
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3BiXZAWk-1689909435268)(https://typora130.oss-cn-nanjing.aliyuncs.com/QQ截图20230511154828.png)] 上述中我们观察到四个线程的PID都是相同的但是LWP是不一样的LWPlightweight process)也就是线程。 监控线程脚本monitorThread.sh):
#! /usr/bin/bash # 1) script description
# use to monitor thread while :;
do ps -aL | head -1 ps -aL | grep thread | grep -v grep; echo -----------------------------------------------------------; sleep 1;
done 2. 线程优缺点
优点
创建一个新线程的代价要比创建一个新进程小得多与进程之间的切换相比线程之间的切换需要操作系统做的工作要少很多线程占用的资源要比进程少很多在等待慢速I/O操作结束的同时程序可执行其他的计算任务计算密集型应用算法相关的消耗CPU资源的为了能在多处理器系统上运行将计算分解到多个线程中实现I/O密集型应用为了提高性能将I/O操作重叠。线程可以同时等待不同的I/O操作
缺点
性能损失。一个很少被外部事件阻塞的计算密集型线程往往无法与共它线程共享同一个处理器。如果计算密集型线程的数量比可用的处理器多那么可能会有较大的性能损失这里的性能损失指的是增加了额外的同步和调度开销而可用的资源不变。健壮性降低。多线程程序中任 何一个线程崩溃最后都会导致进程崩溃缺乏访问控制。
3. 线程使用
3.1 认识线程库 操作系统中没有真正意义的线程只有进程模拟的线程所以Linux操作系统不会提供相关操作线程的系统调用只会提供相关线程的接口那么我们就需要用一个线程库对下来对Linux接口提供封装对上给用户提供线程操作接口和线程管理这个库就是pthread库。通过命令ls /lib64/libpthread.so.* -al即可查看pthread库。 3.2 使用
3.2.1 线程创建
选项内容函数体int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg);参数说明thread:线程id(pthread_t类型就是unsigned long int类型),start_routine:执行函数地址,arg:start_routine执行函数唯一参数,attr:线程属性结构体指针,为NULL时默认创建线程属性返回值成功返回0失败返回相应的错误码
#include iostream
#include pthread.h
#include unistd.h
using namespace std;void* running(void* args)
{while(true){std::cout new thread run... std::endl;sleep(1);}return nullptr;
}int main()
{pthread_t p;pthread_create(p, nullptr, running, nullptr);while(true){std::cout main thread run... new thread id: p std::endl;sleep(1);}return 0;
}运行结果 请问这里的打印的线程id和LWP(lightweight process)的id有什么区别呢pthread_t就是一个包含线程属性的地址数据LWP就是内核中轻量级进程的ID类似于进程ID #include iostream
#include pthread.h
#include cstdio
#include unistd.h
using namespace std;#define NUM 10void* running(void* args)
{char* name (char*)args;while(true){printf(name:%s run...\n, name);sleep(1);}return nullptr;
}int main()
{pthread_t ptable[NUM];for(size_t i 0; i NUM; i){char name[64];snprintf(name, 64, thread-%d, i 1);printf(main:%p\n, name);pthread_create(ptablei, nullptr, running, name); }while(true){std::cout main thread run... new thread id: std::endl;sleep(1);}return 0;
}运行结果 请问为什么后面的打印结果都是一样的而不是从1-10线程编号依次打印呢 因为这里的main函数体中的name这个指针是在栈上的同一进程下创建的线程都是轻量级进程都是同一块的虚拟内存空间所以访问的是用一块栈空间这里的char name[64]的地址是不变的每次snprintf写入数据都是对上次的覆盖所以先前打印的是按照顺序来的但是后面最终char name[64]被覆盖成thread-10后就不会改变了所以最终都是打印一样的。这里我们说访问的是一样的虚拟内存空间那么我们开辟到堆上来试试 #include iostream
#include pthread.h
#include cstdio
#include unistd.h
using namespace std;#define NUM 10void* running(void* args)
{char* name (char*)args;while(true){printf(name:%s run...\n, name);sleep(1);}delete[] name;return nullptr;
}int main()
{pthread_t ptable[NUM];for(size_t i 0; i NUM; i){char* name new char[64];snprintf(name, 64, thread-%d, i 1);pthread_create(ptablei, nullptr, running, name); }while(true){printf(main thread run...\n);sleep(1);}return 0;
}运行结果 为什么这样就可以得到不同线程编号呢? 原因是每次new出来的堆空间都是不同的,都是新的空间,所以每次传给新线程的堆空间都是不同的,snprintf函数写入内容也是不同的,至此打印不同的线程编号.补充问题:为什么这里线程编号依次从小到大排序的呢,原因就是CPU调度器调度谁是不一定的. #include iostream
#include pthread.h
#include cstdio
#include unistd.h
using namespace std;#define NUM 10void* running(void* args)
{char* name (char*)args;while(true){printf(name:%s run...\n, name);sleep(1);}delete[] name;return nullptr;
}int main()
{pthread_t ptable[NUM];for(size_t i 0; i NUM; i){char* name new char[64];snprintf(name, 64, thread-%d, i 1);pthread_create(ptablei, nullptr, running, name); }sleep(2);return 0;
}运行结果 3.2.2 线程等待 现象:程序运行两秒,主线程退出,所有线程退出,这里导致的就类似于僵尸进程,这里并没有僵尸线程概念,那么这里怎么来等待和回收线程呢? 选项内容函数体int pthread_join(pthread_t thread, void **retval);参数说明thread:线程id,retval:执行函数返回值函数功能等待线程返回值成功返回0,失败返回对应的错误码
#include iostream
#include pthread.h
#include cstdio
#include unistd.h
using namespace std;#define NUM 10void* running(void* args) /*可重入函数*/
{char* name (char*)args;while(true){printf(name:%s run...\n, name);sleep(2);break; /*线程退出*/}delete[] name;return nullptr;
}int main()
{pthread_t ptable[NUM];for(size_t i 0; i NUM; i) /*线程创建*/{char* name new char[64];snprintf(name, 64, thread-%d, i 1);pthread_create(ptablei, nullptr, running, name); }for(size_t i 0; i NUM; i) /*线程等待*/{int ret pthread_join(ptable[i], nullptr);if(ret ! 0) cout pthread_join err endl;}cout all thread quit endl;sleep(2);return 0;
}运行结果 3.2.3 线程退出
选项内容函数体void pthread_exit(void *retval);参数说明retval:执行函数返回值函数功能终止线程
#include iostream
#include pthread.h
#include cstdio
#include unistd.h
#include string
using namespace std;#define NUM 10enum {NO, OK};struct threadData /*线程等待返回线程数据*/
{ string _name;int _id;int _retStatus; /*返回状态*/threadData(const string name, const int id):_name(name),_id(id),_retStatus(OK){}
};void* running(void* args) /*可重入函数*/
{threadData* td static_castthreadData*(args);while(true){printf(name:%s, id:%d\n, td-_name.c_str(), td-_id);sleep(2);break;}pthread_exit(td); /*终止线程*/
}int main()
{pthread_t ptable[NUM];for(size_t i 0; i NUM; i) /*线程创建*/{char name[64];snprintf(name, 64, thread-%d, i 1);threadData* td new threadData(name, i1); /*创建线程信息*/pthread_create(ptablei, nullptr, running, td); }void* retval nullptr;for(size_t i 0; i NUM; i) /*线程等待*/{int ret pthread_join(ptable[i], retval);if(ret ! 0) cout pthread_join err endl;threadData* td static_castthreadData*(retval);cout thread exit status: td-_retStatus endl;}cout all thread quit endl;return 0;
}/*为什么没有所谓的异常退出呢*/
/*线程异常进程会退出,进程是承担分配系统资源的基本实体,进程结束对应的资源全部释放*/运行结果 3.2.4 线程取消
选项内容函数体int pthread_cancel(pthread_t thread);函数功能取消线程返回值成功返回0,失败对应的错误码
#include iostream
#include pthread.h
#include cstdio
#include unistd.h
#include string
using namespace std;/*线程取消*/void* running(void* args)
{char* name (char*)args;int cnt 5;while(cnt){cout name is running .. cnt-- endl;sleep(1);}pthread_exit((void*)10);
}int main()
{pthread_t p;pthread_create(p, nullptr, running, (void*)thread-1); /*线程创建*/sleep(3);pthread_cancel(p); /*取消线程*/void* retVal nullptr;pthread_join(p, retVal); /*线程等待*/cout thread-1 exit: (int64_t)retVal endl;return 0;
}运行结果 3.2.5 获取线程id
选项内容函数体pthread_t pthread_self(void);函数功能获取线程id
3.2.6 线程分离
选项内容函数体int pthread_detach(pthread_t thread);函数功能分离线程返回值成功返回0失败返回对应的错误码
#include iostream
#include pthread.h
#include cstring
#include string
#include unistd.hvoid* running(void* args)
{std::string name static_castconst char*(args);int cnt 5;while(cnt){std::cout name : cnt-- std::endl;sleep(1);}return nullptr;
}int main()
{pthread_t p;pthread_create(p, nullptr, running, (void*)thread-1);pthread_detach(p);int ret pthread_join(p, nullptr);if(0 ! ret) std::cerr error: ret : strerror(ret) std::endl;return 0;
}
//输出结果Invaild argument分离的一个线程不能再等待 #include iostream
#include pthread.h
#include cstring
#include string
#include unistd.hvoid* running(void* args)
{pthread_detach(pthread_self());std::string name static_castconst char*(args);int cnt 5;while(cnt){std::cout name : cnt-- std::endl;sleep(1);}return nullptr;
}int main()
{pthread_t p;pthread_create(p, nullptr, running, (void*)thread-1);sleep(1); /*不加上这段代码:正常执行加上这段代码:报错进程自动终止*/int ret pthread_join(p, nullptr);if(0 ! ret) std::cerr error: ret : strerror(ret) std::endl;return 0;
}加和不加上面代码中的sleep(1)会出现不同情况的原因加上sleep(1)可能主线程优先调度那么分离不能等待就会导致报错。本质还是调度优先级问题 3.3 理解线程库 线程库文件是放在硬盘上面的程序运行变成进程此时创建线程后有对应的轻量级进程因为操作系统没有提供相关的系统调用接口那么就需要实现一个用来提供给用户各种线程操作接口和管理线程的库这里叫做pthread库动态库是放在虚拟地址空间的共享区的此时动态库就别包括在进程中那么在内核中就可以获取相应的有关数据就可以对线程做管理并且提供相关的操作接口。这里的pthread_t其实就是线程id它的本质是一个标志线程相关的属性的地址数据只不过我们打印看到的是以十进制方式打印的。所有进程都有自己独立的栈结构主线程是系统栈(虚拟内存空间中的栈)新线程是库中提供的栈(库中自我独立的栈) #include iostream
#include pthread.h
#include cstring
#include string
#include unistd.hstd::string sixteenPrint(pthread_t id)
{char buf[64];snprintf(buf, 64, 0x%x, id);return buf;
}void* running(void* args)
{std::string name static_castconst char*(args);int cnt 5;while(cnt){std::cout name : cnt-- : sixteenPrint(pthread_self()) std::endl;sleep(1);}return nullptr;
}int main()
{pthread_t p;pthread_create(p, nullptr, running, (void*)thread-1);while(true){std::cout main thread: sixteenPrint(pthread_self()) new thread: sixteenPrint(p) std::endl;sleep(1);}int ret pthread_join(p, nullptr);if(0 ! ret) std::cerr error: ret : strerror(ret) std::endl;return 0;
}运行结果 任何语言在Linux下运行底层用的都是pthread库 3.4 证明线程栈
#include iostream
#include pthread.h
#include cstring
#include string
#include unistd.hstd::string sixteenPrint(pthread_t id)
{char buf[64];snprintf(buf, 64, 0x%x, id);return buf;
}void* running(void* args)
{std::string name static_castconst char*(args);int cnt 5; /*每个线程都有自己的cnt*/while(cnt){printf(name:%s, cnt:%d, cnt:%p, newThreadID:%p\n, name.c_str(), cnt--, cnt, sixteenPrint(pthread_self()).c_str());sleep(1);}return nullptr;
}int main()
{pthread_t p1, p2, p3;pthread_create(p1, nullptr, running, (void*)thread-1);pthread_create(p2, nullptr, running, (void*)thread-2);pthread_create(p3, nullptr, running, (void*)thread-3);pthread_join(p1, nullptr);pthread_join(p2, nullptr);pthread_join(p3, nullptr);return 0;
}运行结果 此时我们每个线程内部都有自己独立的栈结构所以cnt的地址是不同的。不是进程地址空间中只有一个栈吗主线程用的是进程系统栈新线程用的是库中提供的栈。 3.5 线程局部存储
#include iostream
#include pthread.h
#include cstring
#include string
#include unistd.hint g_val 10;std::string sixteenPrint(pthread_t id)
{char buf[64];snprintf(buf, 64, 0x%x, id);return buf;
}void* running(void* args)
{std::string name static_castconst char*(args);while(true){printf(g_val:%d, g_val:%p\n, g_val, g_val);sleep(1);}return nullptr;
}int main()
{pthread_t p1, p2, p3;pthread_create(p1, nullptr, running, (void*)thread-1);pthread_create(p2, nullptr, running, (void*)thread-2);pthread_create(p3, nullptr, running, (void*)thread-3);pthread_join(p1, nullptr);pthread_join(p2, nullptr);pthread_join(p3, nullptr);return 0;
}运行结果 解释全局变量是在虚拟内存空间的初始化数据段的 所以地址是不变的对g_val线程得到的值发生变化是因为写时拷贝。那么这里怎么样不让所有线程共享呢定义g_val前加上__thread编译选项 #include iostream
#include pthread.h
#include cstring
#include string
#include unistd.h__thread int g_val 10;std::string sixteenPrint(pthread_t id)
{char buf[64];snprintf(buf, 64, 0x%x, id);return buf;
}void* running(void* args)
{std::string name static_castconst char*(args);while(true){printf(g_val:%d, g_val:%p\n, g_val, g_val);sleep(1);}return nullptr;
}int main()
{pthread_t p1, p2, p3;pthread_create(p1, nullptr, running, (void*)thread-1);pthread_create(p2, nullptr, running, (void*)thread-2);pthread_create(p3, nullptr, running, (void*)thread-3);pthread_join(p1, nullptr);pthread_join(p2, nullptr);pthread_join(p3, nullptr);return 0;
}运行结果 g_val的地址不一样每个线程各自有一份g_val变量。这就是线程局部存储。 4. 线程互斥 线程中大部分资源都是直接或间接共享的这就会导致多线程并发访问的问题为了解决并发访问资源问题线程就有了互斥策略。 4.1 相关概念
临界资源多线程执行流共享的资源就叫做临界资源临界区每个线程内部访问临界资源的代码就叫做临界区不访问临界资源的区域就叫做非临界区互斥任何时刻互斥保证有且只有一个执行流进入临界区访问临界资源通常对临界资源起保护作用原子性不会被任何调度机制打断的操作该操作只有两态要么完成要么未完成
4.2 互斥场景 threadA:while(90){g_val–} threadB:while(10){g_val–} 先threadB运行执行2次后100-98threadA运行发生线程上下文切换threadA运行90次后100-10,threadB运行发生线程上下文切换此时Data cache中存放的是98返回计算值后内存中的g_val变为98原本threadA执行完后g_val变为10但是threadB一执行g_val变为了98最终变为90 4.3 代码演示模拟抢票
#include iostream
#include pthread.h
#include unistd.h
#include stringint tickets 10;void *running(void *name)
{std::string thread_name static_castconst char *(name);while (true){if (tickets 0){usleep(1000); /*模拟抢票花费时间*/printf(thread_name:%s, tickets:%d\n,thread_name.c_str(), tickets--);}else{break;}}return nullptr;
}int main()
{pthread_t thread[4]; // 4个线程int N sizeof(thread) / sizeof(thread[0]);for (int i 0; i N; i) /*创建线程*/{char *name new char[64];snprintf(name, 64, thread-%d, i 1);pthread_create(thread i, nullptr, running, name);}for (int i 0; i N; i) /*等待回收线程*/{pthread_join(thread[i], nullptr);}return 0;
}运行结果 票数能抢到负数吗这里就有问题这里–操作并没有对资源做保护也就是上述互斥场景中的一样是非原子性的线程执行会被任何调度机制打断。为了解决原子性问题这里就需要加锁保护共享资源。 认识锁接口
选项内容初始化int pthread_mutex_destroy(pthread_mutex_t *mutex);销毁int pthread_mutex_init(pthread_mutex_t *restrict mutex, const pthread_mutexattr_t *restrict attr);加锁int pthread_mutex_lock(pthread_mutex_t *mutex);解锁int pthread_mutex_unlock(pthread_mutex_t *mutex);
//加锁保护资源
#include iostream
#include pthread.h
#include unistd.h
#include stringint tickets 10; //票数
pthread_mutex_t mutex; //锁void *running(void *name)
{std::string thread_name static_castconst char *(name);while (true){pthread_mutex_lock(mutex); //加锁成功执行临界区代码失败阻塞当前执行流if (tickets 0){usleep(1000); /*模拟抢票花费时间*/printf(thread_name:%s, tickets:%d\n, \thread_name.c_str(), tickets--);pthread_mutex_unlock(mutex); //解锁}else{pthread_mutex_unlock(mutex); //解锁break;}//TODO:完成后续操作usleep(100);}return nullptr;
}int main()
{pthread_t thread[4]; // 4个线程pthread_mutex_init(mutex, nullptr); //初始化锁int N sizeof(thread) / sizeof(thread[0]);for (int i 0; i N; i) /*创建线程*/{char *name new char[64];snprintf(name, 64, thread-%d, i 1);pthread_create(thread i, nullptr, running, name);}for (int i 0; i N; i) /*等待回收线程*/{pthread_join(thread[i], nullptr);}pthread_mutex_destroy(mutex); //销毁锁return 0;
}运行结果 封装处理
#include iostream
#include pthread.h
#include unistd.h
#include stringint tickets 10;struct thread_info
{std::string _name; /*线程名*/pthread_mutex_t* _mutex_ptr; /*锁地址*/thread_info(const std::string name, pthread_mutex_t* mutex_ptr):_name(name),_mutex_ptr(mutex_ptr){}~thread_info(){}
};void* running(void *args)
{thread_info* info static_castthread_info*(args);while (true){pthread_mutex_lock(info-_mutex_ptr); //加锁成功执行临界区代码失败阻塞当前执行流if (tickets 0){usleep(1000); /*模拟抢票花费时间*/printf(thread_name:%s, tickets:%d\n, \info-_name.c_str(), tickets--);pthread_mutex_unlock(info-_mutex_ptr); //解锁}else{pthread_mutex_unlock(info-_mutex_ptr); //解锁break;}//TODO:完成后续操作usleep(100);}return nullptr;
}int main()
{pthread_t thread[4]; // 4个线程pthread_mutex_t mutex; //锁pthread_mutex_init(mutex, nullptr); //初始化锁int N sizeof(thread) / sizeof(thread[0]);for (int i 0; i N; i) /*创建线程*/{char name[64];snprintf(name, 64, thread-%d, i 1);thread_info* info new thread_info(name, mutex);pthread_create(thread i, nullptr, running, info);}for (int i 0; i N; i) /*等待回收线程*/{pthread_join(thread[i], nullptr);}pthread_mutex_destroy(mutex); //销毁锁return 0;
}4.4 细节
凡是访问同一个临界资源的线程都要进行加锁保护而且必须是加的是同一把锁原因是所有线程都是访问的同一个临界资源都是会互相制约加锁的本质就是对临界区加锁线程访问临界区资源首先需要加锁所有线程都是看到的同一把锁这导致锁本身就是公共资源锁是如何保证自己的安全加锁和解锁本身就是原子性的临界区可能是一行代码也可能是多行代码在临界区代码执行时线程可能被切换吗可能临界区代码和普通代码一样并没有特殊化那么线程在临界区代码执行时切换会受影响吗不会有影响因为此进程把锁拿走了任何线程切换进来后都无法成功申请到锁
4.5 锁的实现原理 代码必须要有互斥行为当代码进入临界区执行时不允许其他线程进入该临界区。如果多个线程同时要求执行临界区的代码并且临界区没有线程在执行那么只能允许一个线程进入该临界区。如果线程不在临界区中执行那么该线程不能阻止其他线程进入临界区。要做到这三点本质上就是需要一把锁。Linux上提供的这把锁叫互斥量。、 为了实现互斥锁操作,大多数体系结构都提供了swap或exchange指令,该指令的作用是把寄存器和内存单元的数据相交换,由于只有一条指令,保证了原子性,即使是多处理器平台,访问内存的 总线周期也有先后,一个处理器上的交换指令执行时另一个处理器的交换指令只能等待总线周期。 加锁和解锁就是变量数据1在内存和CPU寄存器中的流转。这里的1并不是每个线程都有一个这个1只有一个每个线程都是通过swap和exchange方法来得到1完成对应的加锁和解锁操作。这里的锁对应的就是一条汇编指令要么执行完成要么没执行不完成也就保证了原子性。 4.6 线程封装
thread_library.hpp
#pragma once#include iostream
#include string
#include pthread.hnamespace MyThread
{class thread{private:typedef void (*task)(void*);typedef enum {CONSTRUCT, //新建状态RUNNING, //运行状态EXITED //退出状态}status; private:pthread_t _id; //线程idstd::string _name; //线程名task _task; //任务(需被设置)status _status; //状态void* _args; //参数public:thread(int id, task task, void* args):_status(CONSTRUCT),_id(id),_task(task),_args(args){char name[64];snprintf(name, 64, thread-%d, id);_name name;}~thread() {}public:const int getStatus() const {return _status;}const std::string getName() const {return _name;}const pthread_t getId() const {if(_status RUNNING) return _id;else std::cout thread is not running, no id std::endl ; return 0;}void operator()() { _task(_args); } //成员函数调用类中的被设置的任务//类的成员函数具有默认参数this// void* routine(void* args) errorstatic void* routine(void* args) //OK{//_task(_args); //err:static成员函数无法直接访问类内成员 解决方式pthread_create传参数直接传thisthread* self_this static_castthread*(args);(*self_this)(); //对象调用operator()return nullptr;}void run(){int ret pthread_create(_id, nullptr, routine, this);if(ret ! 0) exit(1);_status RUNNING;}void join(){int ret pthread_join(_id, nullptr);if(ret ! 0) {std::cerr join thread: _name (error) std::endl; return;}_id 0;_status EXITED;}}; //!thread} //!MyThreadlock_guard.hpp
#pragma once#include iostream
#include pthread.hnamespace MyThread
{class mutex //对锁封装{private:pthread_mutex_t* _mutex;public:mutex(pthread_mutex_t* mutex):_mutex(mutex){}~mutex() { }public:void lock(){pthread_mutex_lock(_mutex);}void unlock(){pthread_mutex_unlock(_mutex);}}; //!mutex class lockguard //锁保护装置调用对象构造加锁析构解锁{private:mutex _mutex;public:lockguard(pthread_mutex_t* mutex):_mutex(mutex){_mutex.lock();}~lockguard() { _mutex.unlock(); }}; //!lockguard}; //!MyThreadthread_library.cc
#include thread_library.hpp
#include lock_guard.hpp
#include unistd.h//测试封装锁版本
int tickets 10000;pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER; //全局定义锁void task(void *args)
{std::string message static_castconst char*(args);while (true){{ //临界区MyThread::lockguard lockguard(mutex); //创建时自动加锁临时对象出作用域自动解锁if (tickets 0){usleep(1000); /*模拟抢票花费时间*/std::cout message get a ticket: tickets-- std::endl;}else break;}}
}int main()
{MyThread::thread thread1(1, task, (void*)hello1);MyThread::thread thread2(2, task, (void*)hello2);MyThread::thread thread3(3, task, (void*)hello3);thread1.run();thread2.run();thread3.run();thread1.join();thread2.join();thread3.join();pthread_mutex_destroy(mutex);return 0;
}//测试无锁版本
// void routine(void* args)
// {
// std::string message static_castconst char*(args);
// int count 5;
// while(count--)
// {
// std::cout new thread: message std::endl;
// sleep(1);
// }
// }// int main()
// {
// MyThread::thread thread1(1, routine, (void*)hello!);
// std::cout
// thread name: thread1.getName()
// thread id: thread1.getId()
// thread status: thread1.getStatus() std::endl;
// thread1.run();
// std::cout
// thread name: thread1.getName()
// thread id: thread1.getId()
// thread status: thread1.getStatus() std::endl;
// thread1.join();
// std::cout
// thread name: thread1.getName()
// thread id: thread1.getId()
// thread status: thread1.getStatus() std::endl;// MyThread::thread thread2(2, routine, (void*)hello!);
// std::cout
// thread name: thread2.getName()
// thread id: thread2.getId()
// thread status: thread2.getStatus() std::endl;
// thread2.run();
// std::cout
// thread name: thread2.getName()
// thread id: thread2.getId()
// thread status: thread2.getStatus() std::endl;
// thread2.join();
// std::cout
// thread name: thread2.getName()
// thread id: thread2.getId()
// thread status: thread2.getStatus() std::endl;// return 0;
// }5. 可重入函数和线程安全
5.1 概念 线程安全多个线程并发同一段代码时不会出现不同的结果。常见对全局变量或者静态变量进行操作并且没有锁保护的情况下会出现该问题。 可重入函数同一个函数被不同的执行流调用当前一个流程还没有执行完就有其他的执行流再次进入我们称之为重入。一个函数在重入的情况下运行结果不会出现任何不同或者任何问题则该函数被称为可重入函数否则是不可重入函数。 5.2 常见线程安全情况
每个线程对全局变量或者静态变量只有读取的权限而没有写入的权限一般来说这些线程是安全的类或者接口对于线程来说都是原子操作多个线程之间的切换不会导致该接口的执行结果存在二义性
5.3 常见不可重入情况
调用了malloc/free函数因为malloc函数是用全局链表来管理堆的调用了标准I/O库函数标准I/O库的很多实现都以不可重入的方式使用全局数据结构可重入函数体内使用了静态的数据结构
5.4 常见可重入情况
不使用全局变量或静态变量不使用用malloc或者new开辟出的空间不调用不可重入函数不返回静态或全局数据所有数据都有函数的调用者提供使用本地数据或者通过制作全局数据的本地拷贝来保护全局数据
5.5 可重入函数和线程安全关系
函数是可重入的那就是线程安全的函数是不可重入的那就不能由多个线程使用有可能引发线程安全问题如果一个函数中有全局变量或静态变量那么这个函数既不是线程安全也不是可重入的
5.6 可重入函数和线程安全的区别
线程安全不一定是可重入的而可重入函数则一定是线程安全的如果将对临界资源的访问加上锁则这个函数是线程安全的但如果这个重入函数若锁还未释放则会产生死锁因此是不可重入的
6. 死锁 死锁是指在一组进程中的各个进程均占有不会释放的资源但因互相申请被其他进程所站用不会释放的资源而处于的一种永久等待状态。 6.1 四个充分必要条件
互斥条件一个资源每次只能被一个执行流使用请求与保持条件一个执行流因请求资源而阻塞时对已获得的资源保持不放循环等待条件:若干执行流之间形成一种头尾相接的循环等待资源的关系不剥夺条件:一个执行流已获得的资源在末使用完之前不能强行剥夺
6.2 避免死锁 上述四个充分必要条件只要有一个不满足就是死锁不满足第一个条件的做法不加锁不满足第二个条件的做法不主动释放锁不满足第三个条件的做法不按照顺序申请锁不满足第四个条件的做法不控制线程统一释放锁。 7. 线程同步
7.1 概念 饥饿问题的产生张三是103自习室的钥匙掌管者他每次去103自习室自习都有这样一个毛病每次出去上厕所或者买饭什么的他都把钥匙带在身上并不把钥匙挂门上防止其他人趁他不在自习室占位子这让导致后面等待自习的同学就无法拿到钥匙这就是饥饿问题。发生这种现象有的同学就去管理员举报张三管理员最终给了个规则自习完毕或者出自习室的人都要把钥匙挂在门上给其他等待同学自习自习完毕的人不能立即申请钥匙并且等待自习的人要排队。这就是同步。同步就是在保证数据安全的前提下让线程能够按照某种特定的顺序访问临界资源从而有效避免饥饿问题 7.2 条件变量 条件变量是利用线程间共享的全局变量进行同步的一种机制主要包括两个动作一个线程等待“条件变量的条件成立”而挂起另一个线程使“条件成立”给出条件成立信号 7.2.1 条件变量初始化
选项内容函数体int pthread_cond_init(pthread_cond_t *restrict cond, const pthread_condattr_t *restrict attr);函数功能初始化条件变量返回值成功返回0失败返回错误码
7.2.2 条件变量销毁
选项内容函数体int pthread_cond_destroy(pthread_cond_t *cond);函数功能销毁条件变量返回值成功返回0失败返回错误码
7.2.3 条件变量等待
选项内容函数体int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);函数功能等待条件变量返回值成功返回0失败返回错误码
7.2.4 唤醒等待
选项内容函数体int pthread_cond_signal(pthread_cond_t *cond); //单独唤醒 int pthread_cond_broadcast(pthread_cond_t *cond); //全部唤醒函数功能唤醒条件变量返回值成功返回0失败返回错误码
7.3 demo
#include iostream
#include pthread.h
#include cstdio
#include unistd.h
using namespace std;#define NUM 5//全局初始化
pthread_cond_t condition_variable PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex PTHREAD_MUTEX_INITIALIZER;void* active(void* args)
{string name static_castconst char*(args);while(true){pthread_mutex_lock(mutex); //加锁pthread_cond_wait(condition_variable, mutex); //线程挂起等待(调用时会自动释放锁)cout name active endl;pthread_mutex_unlock(mutex);}
}int main()
{pthread_t threads[NUM]; //NUM个线程for(size_t i 0; i NUM; i){char* name new char[32];snprintf(name, 32, thread-%d, i1);pthread_create(threads[i], nullptr, active, name);}sleep(2); //等待2秒后线程被随机唤醒while(true){cout thread waked up endl; pthread_cond_signal(condition_variable);sleep(1);}for(size_t i 0; i NUM; i){pthread_join(threads[i], nullptr);}return 0;
}8. 生产者-消费者模型
8.1 概念 提到生产者和消费者就想到了买东西我们从买东西说起往往我们需要什么的时候我们就会去超市购买我们是消费者生产者其实就是超市的供货商那么超市是个什么呢超市既不是消费者也不是生产者而是交易场所。供货商可以源源不断的生产消费者源源不断的购买超市也就承载了商品大量输入输出的工作也就提高了效率。生产者生产和消费者消费可以不同时进行可能生产者看到超市中的商品还有很多也有库存此时生产者就没有生产这类商品了但是消费者依旧每天都在购买还有一种情况就是这类商品没有了而消费者想买但没有此时消费者就无法购买而不再购买此时生产者看到商品没有了也就要加力生产。这也就体现了生产消费可以同时进行忙闲不均。计算机中生产者和消费者都是线程超市也就是特定的缓冲区超市内的商品就是数据。超市这个交易场所能被支持消费者消费和生产者生产的前提就是被生产者和消费者可进出。也就是这个特定的缓冲区被所有线程先看到这个特定的缓冲区才能被使用那么这个特定的缓冲区就是共享资源。 8.2 优点和关系 优点解耦支持并发支持忙闲不均 关系生产者和生产者是互斥关系生产者和消费者是同步关系和互斥关系消费者和消费者是互斥关系 口诀三种关系、两种角色(生产者和消费者)、一个交易场所(缓冲区) 8.3 BlockQueue生产者-消费者模型
blockqueue.hpp
#pragma once
#include task.hpp
#include iostream
#include queue
#include pthread.h//队列满了不能生产了应该让消费者消费生产者等待
//队列空了不能消费了应该让生产者生产消费者等待template class T
class blockqueue
{
private:std::queueT _queue; //阻塞队列int _capacity; //容量pthread_mutex_t _mutex; //锁(一把锁保证线程同步访问共享资源)pthread_cond_t _producer_cond; //控制生产者生产或等待条件变量pthread_cond_t _consumer_cond; //控制消费者消费或等待条件变量
public:blockqueue(int capacity 5):_capacity(capacity){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_producer_cond, nullptr);pthread_cond_init(_consumer_cond, nullptr);}~blockqueue() {pthread_mutex_destroy(_mutex);pthread_cond_destroy(_producer_cond);pthread_cond_destroy(_consumer_cond);}
private:bool isfull() const { return _queue.size() _capacity; }bool isempty() const { return _queue.empty(); }
public:void push(const T data) //队列满了生产等待唤醒消费者消费{pthread_mutex_lock(_mutex);while(isfull()) //if(isfull())可能存在多线程误唤醒继续执行if判断条件后代码从而导致队列中数据超出容量大小问题 - 解决:while(isfull()){//为什么pthread_cond_wait要带上锁参数因为线程等待持有锁会导致饥饿问题因此必须传锁参数释放锁//再次被唤醒后从哪里运行因为是在临界区中加上条件变量来等待所以被唤醒后也是从pthread_cond_wait等待处向后运行pthread_cond_wait(_producer_cond, _mutex); }_queue.push(data);pthread_cond_signal(_consumer_cond); //唤醒消费者pthread_mutex_unlock(_mutex);}void pop(T* data) //队列空了消费者等待唤醒生产者生产{pthread_mutex_lock(_mutex);while(isempty()) //同理while(isfull()){pthread_cond_wait(_consumer_cond, _mutex); }*data _queue.front();_queue.pop();pthread_cond_signal(_producer_cond); //唤醒生产者pthread_mutex_unlock(_mutex);}}; //!blockqueuetask.hpp
#pragma once
#include iostream
#include string//两数加减乘除任务class task
{
private:int _x; //左操作数int _y; //右操作数int _result; //结果char _operation; //操作public:task() {}task(const int x, const int y, const char operation):_x(x),_y(y),_operation(operation),_result(0){}~task() {}public:void operator()(){switch(_operation){case :_result _x _y;break;case -:_result _x - _y;break;case *:_result _x * _y;break;case /:if(_y 0) { std::cout division by zero err std::endl; break; }else _result _x / _y;break;case %:_result _x % _y;break;default:break;}}std::string printresult(){return std::to_string(_result);}std::string printargs(){return std::to_string(_x) _operation std::to_string(_y) ?;}}; //!tasktest.cc
#include block_queue.hpp
#include task.hpp
#include pthread.h
#include ctime
#include cstdlib
#include unistd.hvoid* consumer_execute(void* args)
{blockqueuetask* bq static_castblockqueuetask*(args);while(true){//sleep(1);//从阻塞队列拉取数据task t;bq-pop(t);//通过数据处理相关业务t(); //调用operator()()std::cout pthread_self() pop result : t.printresult() std::endl;}
}void* producer_execute(void* args)
{blockqueuetask* bq static_castblockqueuetask*(args);std::string operations -*/%;while(true){//sleep(1);//生产数据int x rand() % 10 1;int y rand() % 15 1;char operation operations[rand() % (operations.size())];//数据推送到阻塞队列(共享资源)task t(x, y, operation);bq-push(t);std::cout pthread_self() push task: t.printargs() std::endl;}
}int main()
{srand((size_t)time(nullptr));blockqueuetask* bq new blockqueuetask(); //阻塞队列(共享资源)pthread_t consumer[4], producer[2];pthread_create(producer[0], nullptr, producer_execute, bq); //生产者线程pthread_create(producer[1], nullptr, producer_execute, bq); pthread_create(consumer[0], nullptr, consumer_execute, bq); //消费者线程pthread_create(consumer[1], nullptr, consumer_execute, bq);pthread_create(consumer[2], nullptr, consumer_execute, bq);pthread_create(consumer[3], nullptr, consumer_execute, bq);pthread_join(producer[0], nullptr);pthread_join(producer[1], nullptr);pthread_join(consumer[0], nullptr);pthread_join(consumer[1], nullptr);pthread_join(consumer[2], nullptr);pthread_join(consumer[3], nullptr);delete bq;bq nullptr;return 0;
}8.4 POSIX信号量 POSIX信号量和SystemV信号量作用相同都是用于同步操作达到无冲突的访问共享资源目的。 但POSIX可以用于线程间同步。本质就是一个计数器用来描述多线程共享资源的多少申请信号量成功便有对应资源可以使用反之不成功对应资源不可用进行PV操作P操作wait操作被称为ProbeerV操作signal操作被称为Verhoog。 8.4.1 信号量初始化
选项内容函数体int sem_init(sem_t *sem, int pshared, unsigned int value);函数功能信号量初始化pshared:0表示线程间共享非零表示进程间共享value信号量初始值返回值成功返回0失败返回-1错误码被设置
8.4.2 信号量销毁
选项内容函数体int sem_destroy(sem_t *sem);函数功能销毁信号量返回值成功返回0失败返回-1错误码被设置
8.4.3 等待信号量
选项内容函数体int sem_wait(sem_t *sem);函数功能等待信号量信号量值-1返回值成功返回0失败返回-1错误码被设置
8.4.4 发布信号量
选项内容函数体int sem_post(sem_t *sem);函数功能发布信号量信号量值1返回值成功返回0失败返回-1错误码被设置
8.5 RingQueue生产者-消费者模型 利用数组模拟环形队列生产者在队列尾部push数据消费者在队列头部pop数据满和空的判断生产者和消费者指向同一数组空间时队列就是满的生产者和消费者不指向同一数组空间时队列就是空的 只有空和满的时候生产者和消费者是访问同一个位置的不为空和满的时候生产者和消费者是可以并发访问的资源的。另外生产者关注的是空间消费者关心的是数据那么生产者初始信号量就是数组大小而消费者初始信号量就是0没有数据时生产者先生产生产者就P操作申请空间信号量数据放进去此时生产者完成V操作释放数据信号量当有数据信号量时消费者就P操作申请数据信号量空间就释放处一个此时消费者就V操作释放空间信号量。 ring_queue.hpp
#pragma once
#include iostream
#include semaphore.h
#include vector
#include pthread.hstatic const size_t N 5;template class T
class ringqueue
{
private:std::vectorT _queue; //环形队列int _capacity; //容量int _consumerindex; //消费位置int _producerindex; //生产位置sem_t _datasem; //数据信号量(仅消费者关心)sem_t _spacesem; //空间信号量(仅生产者关心)pthread_mutex_t _consumermutex; //消费者之间的锁(保证消费者之间的同步)pthread_mutex_t _producermutex; //生产者之间的锁(保证生产者之间的同步)public:ringqueue(int num N):_queue(num),_capacity(num){sem_init(_datasem, 0, 0); //初始没有数据sem_init(_spacesem, 0, _capacity); //初始空间全部未使用pthread_mutex_init(_consumermutex, nullptr);pthread_mutex_init(_producermutex, nullptr);_producerindex _consumerindex 0;}~ringqueue(){sem_destroy(_datasem);sem_destroy(_spacesem);pthread_mutex_destroy(_consumermutex);pthread_mutex_destroy(_producermutex);}public://生产者先申请锁还是先申请信号量//先申请信号量(先分配空间资源再分配锁好比先占好位置再上厕所(此时空间资源先得到上锁后去上厕所别人就无法占用相比先申请锁效率略高)void push(const T data){P(_spacesem); //P操作:申请空间信号量(一定有空间资源无需判断)lock(_producermutex);_queue[_producerindex] data;_producerindex % _capacity;V(_datasem); //V操作:释放数据信号量unlock(_producermutex);}void pop(T* data){P(_datasem); //P操作:申请数据信号量(一定有数据无需判断)lock(_consumermutex); //和push()中同理(*data) _queue[_consumerindex];_consumerindex % _capacity;V(_spacesem); //V操作:释放空间信号量unlock(_consumermutex);}private:void P(sem_t sem) { sem_wait(sem); } //P:信号量-1void V(sem_t sem) { sem_post(sem); } //V:信号量1void lock(pthread_mutex_t mutex) { pthread_mutex_lock(mutex); } //加锁void unlock(pthread_mutex_t mutex) { pthread_mutex_unlock(mutex); } //解锁}; //!ringqueuetask.hpp
#pragma once
#include iostream
#include string
#include unistd.h//两数加减乘除任务class task
{
private:int _x; //左操作数int _y; //右操作数int _result; //结果char _operation; //操作public:task() {}task(const int x, const int y, const char operation):_x(x),_y(y),_operation(operation),_result(0){}~task() {}public:void operator()(){switch(_operation){case :_result _x _y;break;case -:_result _x - _y;break;case *:_result _x * _y;break;case /:if(_y 0) { std::cout division by zero err std::endl; break; }else _result _x / _y;break;case %:_result _x % _y;break;default:break;}usleep(1000); //处理任务所花费时间}std::string printresult(){return std::to_string(_x) _operation std::to_string(_y) std::to_string(_result);}std::string printargs(){return std::to_string(_x) _operation std::to_string(_y) ?;}}; //!tasktest.cc
#include ring_queue.hpp
#include task.hpp
#include pthread.h
#include ctime
#include unistd.h//线程一定是生产者先运行生产者信号量初始值是N不会等待挂起消费者信号量初始值是0会等待挂起void* producer_execute(void* args)
{ringqueuetask* rq static_castringqueuetask*(args);std::string operations -*/%;while(true){sleep(1);int x rand() % 10 1;int y rand() % 20 1;char operation operations[rand() % operations.size()];task t(x, y, operation);rq-push(t);std::cout pthread_self() push task: t.printargs() std::endl;}
}void* consumer_execute(void* args)
{ringqueuetask* rq static_castringqueuetask*(args);while(true){task t;rq-pop(t);t(); //调用operator()()std::cout pthread_self() pop result : t.printresult() std::endl;}
}//多生产多消费
//加锁的意义防止生产者同时构建任务防止消费者同时处理任务
//信号量的意义不用堆临界区做判断就知道临界资源的使用情况int main()
{srand((size_t)time(nullptr));ringqueuetask* rq new ringqueuetask();pthread_t consumer[4], producer[2];for(size_t i 0; i 2; i){pthread_create(producer i, nullptr, producer_execute, rq);}for(size_t i 0; i 4; i){pthread_create(consumer i, nullptr, consumer_execute, rq);}for(size_t i 0; i 2; i){pthread_join(producer[i], nullptr);}for(size_t i 0; i 4; i){pthread_join(consumer[i], nullptr);}return 0;
}9. 线程池
threadpool_V1.hpp
#pragma once
#include iostream
#include string
#include unistd.h
#include vector
#include queue
#include pthread.hstatic const size_t N 5;//version1: 创建和启动线程同步template class T
class threadpool
{
private:std::vectorpthread_t _threads; //线程池size_t _size; //线程个数std::queueT _tasks; //任务队列pthread_mutex_t _mutex; //保证多线程同步访问任务队列的锁pthread_cond_t _condition; //等待/唤醒线程public:threadpool(int num N):_size(num),_threads(num){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_condition, nullptr);}~threadpool(){pthread_mutex_destroy(_mutex);pthread_cond_destroy(_condition);}public:void init() {}void start(){for(size_t i 0; i _size; i){pthread_create(_threads[i], nullptr, startroutine, this);}}void pushtask(const T t){lock();_tasks.push(t);threadwakeup(); //唤醒线程unlock();}private:static void* startroutine(void* args) //static修饰减少类this指针,无法访问类成员变量{ pthread_detach(pthread_self()); //当一个分离的线程终止时它的资源被自动释放回系统threadpoolT* self_this static_castthreadpoolT*(args);while(true){//检测任务列表中是否有任务有处理无等待self_this-lock();while(self_this-isempty()){self_this-threadwait();}T t self_this-poptask();self_this-unlock();t(); //处理任务是私有并没有访问共享资源无需在临界区中处理std::cout disposed task - result: t.printresult() std::endl;}}void lock() { pthread_mutex_lock(_mutex); }void unlock() { pthread_mutex_unlock(_mutex); }void threadwait() { pthread_cond_wait(_condition, _mutex); }void threadwakeup() { pthread_cond_signal(_condition); }bool isempty() { return _tasks.empty(); }T poptask(){T t _tasks.front();_tasks.pop();return t;}}; //!threadpoolthread_library.hpp
#pragma once#include iostream
#include string
#include pthread.hnamespace MyThread
{class thread{private:typedef void (*task)(void*);typedef enum {CONSTRUCT 0, //新建状态RUNNING, //运行状态EXITED //退出状态}status; private:pthread_t _id; //线程idstd::string _name; //线程名task _task; //任务(需被设置)status _status; //状态void* _args; //参数public:thread(size_t id, task task, void* args):_status(CONSTRUCT),_id(id),_task(task),_args(args){char name[64];snprintf(name, 64, thread-%d, id);_name name;}~thread() {}public:const int getStatus() const {return _status;}const std::string getName() const {return _name;}const pthread_t getId() const {if(_status RUNNING) return _id;else std::cout thread is not running, no id std::endl ; return 0;}void operator()() { if(_args ! nullptr) _task(_args); } //成员函数调用类中的被设置的任务//类的成员函数具有默认参数this// void* routine(void* args) errorstatic void* routine(void* args) //OK{//_task(_args); //err:static成员函数无法直接访问类内成员 解决方式pthread_create传参数直接传thisthread* self_this static_castthread*(args);(*self_this)(); //对象调用operator()return nullptr;}void run(){int ret pthread_create(_id, nullptr, routine, this);if(ret ! 0) exit(1);_status RUNNING;}void join(){int ret pthread_join(_id, nullptr);if(ret ! 0) {std::cerr join thread: _name (error) std::endl; return;}_id 0;_status EXITED;}}; //!thread} //!MyThreadthreadpool_V2.hpp
#pragma once
#include iostream
#include unistd.h
#include vector
#include queue
#include thread_library.hppstatic const size_t N 5;// 引入自定义线程使得创建和启动线程分离template class T
class threadpool
{
private:std::vectorMyThread::thread _threads; //线程池size_t _size; //线程个数std::queueT _tasks; //任务队列pthread_mutex_t _mutex; //保证多线程同步访问任务队列的锁pthread_cond_t _condition; //等待/唤醒线程public:threadpool(int num N):_size(num){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_condition, nullptr);}~threadpool(){for(auto thread : _threads){thread.join();}pthread_mutex_destroy(_mutex);pthread_cond_destroy(_condition);}public:void init() //创建线程{for(size_t i 0; i _size; i){_threads.push_back(MyThread::thread(i, startroutine, this));}}void start() //运行线程{for(auto thread : _threads){thread.run();}}void pushtask(const T t){lock();_tasks.push(t);threadwakeup(); //唤醒线程unlock();}private:static void startroutine(void* args) //static修饰减少类this指针,无法访问类成员变量{ //pthread_detach(pthread_self()); threadpoolT* self_this static_castthreadpoolT*(args);while(true){//检测任务列表中是否有任务有处理无等待self_this-lock();while(self_this-isempty()){self_this-threadwait();}T t self_this-poptask();self_this-unlock();t(); //处理任务是私有并没有访问共享资源无需在临界区中处理std::cout disposed task - result: t.printresult() std::endl;}}void lock() { pthread_mutex_lock(_mutex); }void unlock() { pthread_mutex_unlock(_mutex); }void threadwait() { pthread_cond_wait(_condition, _mutex); }void threadwakeup() { pthread_cond_signal(_condition); }bool isempty() { return _tasks.empty(); }T poptask(){T t _tasks.front();_tasks.pop();return t;}}; //!threadpoollock_guard.hpp
#pragma once#include iostream
#include pthread.h//对锁封装namespace MyThread
{class mutex {private:pthread_mutex_t* _mutex;public:mutex(pthread_mutex_t* mutex):_mutex(mutex){}~mutex() { }public:void lock(){pthread_mutex_lock(_mutex);}void unlock(){pthread_mutex_unlock(_mutex);}}; //!mutex class lockguard //锁保护装置调用对象构造加锁析构解锁{private:mutex _mutex;public:lockguard(pthread_mutex_t* mutex):_mutex(mutex){_mutex.lock();}~lockguard() { _mutex.unlock(); }}; //!lockguard}; //!MyThreadthreadpool_V3.hpp
#pragma once
#include iostream
#include unistd.h
#include vector
#include queue
#include lock_guard.hpp
#include thread_library.hppstatic const size_t N 5;// 引入自定义线程使得创建和启动线程分离 自定义锁重写加锁解锁template class T
class threadpool
{
private:std::vectorMyThread::thread _threads; // 线程池size_t _size; // 线程个数std::queueT _tasks; // 任务队列pthread_mutex_t _mutex; // 保证多线程同步访问任务队列的锁pthread_cond_t _condition; // 等待/唤醒线程public:threadpool(int num N): _size(num){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_condition, nullptr);}~threadpool(){for (auto thread : _threads){thread.join();}pthread_mutex_destroy(_mutex);pthread_cond_destroy(_condition);}public:void init() // 创建线程{for (size_t i 0; i _size; i){_threads.push_back(MyThread::thread(i, startroutine, this));}}void start() // 运行线程{for (auto thread : _threads){thread.run();}}void pushtask(const T t){MyThread::lockguard lock(_mutex);_tasks.push(t);threadwakeup(); // 唤醒线程}private:static void startroutine(void *args) // static修饰减少类this指针,无法访问类成员变量{// pthread_detach(pthread_self());threadpoolT *self_this static_castthreadpoolT *(args);while (true){T t;{ //临界区MyThread::lockguard lock(self_this-getmutex());// 检测任务列表中是否有任务有处理无等待while (self_this-isempty()){self_this-threadwait();}t self_this-poptask();}t(); // 处理任务是私有并没有访问共享资源无需在临界区中处理std::cout disposed task - result: t.printresult() std::endl;}}void lock() { pthread_mutex_lock(_mutex); }void unlock() { pthread_mutex_unlock(_mutex); }void threadwait() { pthread_cond_wait(_condition, _mutex); }void threadwakeup() { pthread_cond_signal(_condition); }bool isempty() { return _tasks.empty(); }pthread_mutex_t *getmutex() { return _mutex; }T poptask(){T t _tasks.front();_tasks.pop();return t;}}; //! threadpooltask.hpp
#pragma once
#include iostream
#include string
#include unistd.h//两数加减乘除任务class task
{
private:int _x; //左操作数int _y; //右操作数int _result; //结果char _operation; //操作public:task() {}task(const int x, const int y, const char operation):_x(x),_y(y),_operation(operation),_result(0){}~task() {}public:void operator()(){switch(_operation){case :_result _x _y;break;case -:_result _x - _y;break;case *:_result _x * _y;break;case /:if(_y 0) { std::cout division by zero err std::endl; break; }else _result _x / _y;break;case %:_result _x % _y;break;default:break;}}std::string printresult(){return std::to_string(_x) _operation std::to_string(_y) std::to_string(_result);}std::string printargs(){return std::to_string(_x) _operation std::to_string(_y) ?;}}; //!tasktest.cc
//#include threadpool_V1.hpp
//#include threadpool_V2.hpp
#include threadpool_V3.hpp
#include task.hpp
#include memoryint main()
{std::unique_ptrthreadpooltask tp(new threadpooltask()); //交给智能指针来管理tp-init();tp-start();while(true){int x, y; char operation;std::cout enter x:;std::cin x;std::cout enter y:;std::cin y;std::cout enter operation(-*/%):;std::cin operation;task t(x, y, operation);tp-pushtask(t);usleep(100);}return 0;
}threadpool_V4.hpp
#pragma once
#include iostream
#include unistd.h
#include vector
#include queue
#include lock_guard.hpp
#include thread_library.hpp// 引入自定义线程使得创建和启动线程分离 自定义锁重写加锁解锁 单例模式static const size_t N 5;template class T
class threadpool
{
private:std::vectorMyThread::thread _threads; // 线程池size_t _size; // 线程个数std::queueT _tasks; // 任务队列pthread_mutex_t _mutex; // 保证多线程同步访问任务队列的锁pthread_cond_t _condition; // 等待/唤醒线程static threadpoolT *_instance; // 单例线程池地址static pthread_mutex_t _instancemutex; // 保证多线程单例模式private:threadpool(int num N): _size(num){pthread_mutex_init(_mutex, nullptr);pthread_cond_init(_condition, nullptr);}threadpool(const threadpool tp) delete;threadpool operator(const threadpool tp) delete;public:// 单次_instancenullptr判断的问题多线程下可能多个线程同时调用getinstance() - 可能会多次new而有多份单例// 解决方法加锁static threadpoolT *getinstance() // 静态成员{if (_instance nullptr) // 提高效率单例模式下只有一次ne防止每次调用getinstance()都加锁{MyThread::lockguard lock(_instancemutex);if (_instance nullptr) // 保证单例模式{_instance new threadpoolT();_instance-init();_instance-start();}}return _instance;}~threadpool(){for (auto thread : _threads){thread.join();}pthread_mutex_destroy(_mutex);pthread_cond_destroy(_condition);}public:void init() // 创建线程{for (size_t i 0; i _size; i){_threads.push_back(MyThread::thread(i, startroutine, this));}}void start() // 运行线程{for (auto thread : _threads){thread.run();}}void pushtask(const T t){MyThread::lockguard lock(_mutex);_tasks.push(t);threadwakeup(); // 唤醒线程}private:static void startroutine(void *args) // static修饰减少类this指针,无法访问类成员变量{// pthread_detach(pthread_self());threadpoolT *self_this static_castthreadpoolT *(args);while (true){T t;{ // 临界区MyThread::lockguard lock(self_this-getmutex());// 检测任务列表中是否有任务有处理无等待while (self_this-isempty()){self_this-threadwait();}t self_this-poptask();}t(); // 处理任务是私有并没有访问共享资源无需在临界区中处理std::cout disposed task - result: t.printresult() std::endl;}}void lock() { pthread_mutex_lock(_mutex); }void unlock() { pthread_mutex_unlock(_mutex); }void threadwait() { pthread_cond_wait(_condition, _mutex); }void threadwakeup() { pthread_cond_signal(_condition); }bool isempty() { return _tasks.empty(); }pthread_mutex_t *getmutex() { return _mutex; }T poptask(){T t _tasks.front();_tasks.pop();return t;}}; //! threadpooltemplate class T
threadpoolT* threadpoolT::_instance nullptr;
template class T
pthread_mutex_t threadpoolT::_instancemutex PTHREAD_MUTEX_INITIALIZER;test_threadpoolV4.cc
#include threadpool_V4.hpp
#include task.hpp
#include memoryint main()
{while(true){int x, y; char operation;std::cout enter x:;std::cin x;std::cout enter y:;std::cin y;std::cout enter operation(-*/%):;std::cin operation;task t(x, y, operation);threadpooltask::getinstance()-pushtask(t);usleep(100);}return 0;
}10. 常见锁 悲观锁在每次取数据时总是担心数据会被其他线程修改所以会在取数据前先加锁读锁写锁行锁等当其他线程想要访问数据时被阻塞挂起。 乐观锁每次取数据时候总是乐观的认为数据不会被其他线程修改因此不上锁。但是在更新数据前会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式版本号机制和CAS操作当需要更新数据时判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败失败则重试一般是一个自旋的过程即不断重试。。 自旋锁线程获取锁的时候如果锁被其他线程持有则当前线程将循环等待直到获取到锁。自旋锁等待期间线程的状态不会改变线程一直是用户态并且是活动的 (active)。自旋锁如果持有锁的时间太长则会导致其它等待获取锁的线程耗尽CPU。自旋锁本身无法保证公平性同时也无法保证可重入性。 读写锁读写锁也称为共享-独占shared-exclusive锁当读写锁以读模式加锁时它是以共享模式锁住当以写模式加锁时它是以独占模式锁住。 读写锁非常适合读数据的频率远大于写数据的频率从的应用中。 这样可以在任何时刻运行多个读线程并发的执行给程序带来了更高的并发度。只要没有写模式下的加锁任意线程都可以进行读模式下的加锁只有读写锁处于不加锁状态时才能进行写模式下的加锁。 自旋锁
int pthread_spin_destroy(pthread_spinlock_t *lock);int pthread_spin_init(pthread_spinlock_t *lock, int pshared);int pthread_spin_lock(pthread_spinlock_t *lock);int pthread_spin_unlock(pthread_spinlock_t *lock);
读写锁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,const pthread_rwlockattr_t * restrict attr);int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
