文章目录
- 1、atomic使用
- 2、volatile关键字
- 3、条件变量
- 4、成员函数指针使用
- 5、线程池
- 6、主线程先退出对子线程影响
- 7、return、exit、pthread_exit区别
- 8、进程和线程的区别
1、atomic使用
原子操作，不可分割的操作，要么完整，要么不完整。
#include <pthread.h>
#include <unistd.h>
#include <iostream>
#include <atomic>
using namespace std;
atomic<int> g_acount;
int g_count = 0;
void* ThreadFunc(void* threadData)
{
for(int i=0;i<1000000;i++)
{
g_count++;
g_acount++;
}
}
int main(int argc, const char** argv)
{
pthread_t pid1,pid2;
int err = pthread_create(&pid1,NULL,ThreadFunc,NULL);
if(err!=0)
{
cout<<"thread fail---"<<endl;
exit(0);
}
err = pthread_create(&pid2,NULL,ThreadFunc,NULL);
if(err!=0)
{
cout<<"thread fail---"<<endl;
exit(0);
}
pthread_join(pid1,NULL);
pthread_join(pid2,NULL);
cout<<"g_count:"<<g_count<<endl;
cout<<"g_acount:"<<g_acount<<endl;
return 0;
}
makefile
all: pthreadText
pthreadText:pthreadText.cpp
g++ -o pthreadText pthreadText.cpp -pthread -std=c++11
运行结果：
2、volatile关键字
用volatile关键字声明的变量，会告诉编译器，这个变量随时可能发生变化，编译器在编译的时候就不会对变量进行激进的优化，每次去读取的时候都会去内存中取，相反，如果编译器进行量优化，可能读取的时候去寄存器去读取这个值，三种特性：易变的、不可优化的、顺序执行的。
3、条件变量
条件本身（while((g_msgQueue.size() == 0) && isRuning == false))是由互斥量保护的，线程在发生改变之前首先锁住互斥量，其他线程不会察觉到这种改变，因为互斥量必须锁住才能计算条件。
#include <iostream>
#include <signal.h>
#include <unistd.h>
#include <pthread.h>
#include <list>
using namespace std;
// 初始化
pthread_cond_t g_cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
struct msgStr
{
char name[256];
int age;
int id;
};
list<msgStr*> g_msgQueue;
bool isRuning = false;
void* outCache(void* data)
{
while(true){
pthread_mutex_lock(&g_mutex);
while((g_msgQueue.size() == 0) && isRuning == false){
pthread_cond_wait(&g_cond,&g_mutex);
}
if(isRuning){
pthread_mutex_unlock(&g_mutex);
break;
}
// 消息处理
msgStr* jobbuf= g_msgQueue.front();
g_msgQueue.pop_front();
pthread_mutex_unlock(&g_mutex);
// 消息处理
cout<<"tid:"<<pthread_self()
<<"name:"<<jobbuf->name
<<"age:"<<jobbuf->age
<<"id:"<<jobbuf->id
<<endl;
usleep(1000);
delete jobbuf;
jobbuf = NULL;
}
}
void inCache(int sig)
{
// 收到15这个信号,向消息队列中添加数据
if(sig == 15){
struct msgStr* msg = NULL;
pthread_mutex_lock(&g_mutex);
for(int i=0;i<1000;i++){
msg = new msgStr();
sprintf(msg->name,"name--%d",i);
msg->age = i;
msg->id = 1000+i;
g_msgQueue.push_back(msg);
}
pthread_mutex_unlock(&g_mutex);
pthread_cond_broadcast(&g_cond);
}
if(sig == 10)
{
isRuning = true;
}
}
int main()
{
// 作为向消息队列中添加数据的函数
signal(15,inCache);
pthread_t pid1,pid2;
pthread_create(&pid1,NULL,outCache,NULL);
pthread_create(&pid2,NULL,outCache,NULL);
pthread_join(pid1,NULL);
pthread_join(pid2,NULL);
return 0;
}
4、成员函数指针使用
#include <iostream>
using namespace std;
class Test
{
public:
Test();
~Test(){}
void func(int a,int b)
{
cout<<"Test:"<<a<<endl;
cout<<"Test:"<<b<<endl;
}
void func1(int a,int b)
{
cout<<"Test:"<<a<<endl;
cout<<"Test:"<<b<<endl;
}
};
// 函数指针
typedef void (Test::*handler)(int a,int b);
const handler handArray[] =
{
NULL,
NULL,
NULL,
NULL,
NULL,
&Test::func1,
&Test::func,
};
Test::Test()
{
(this->*handArray[5])(1,2);
}
int main()
{
Test t;
(t.*handArray[6])(3,5);
return 0;
}
makefile
g++ -o main pthreadPoolText.cpp
5、线程池
线程池概率：提前创建多个线程，并通过一个类来统一管理这一堆线程。
工作流程：来了一个任务，从线程池中找一个空闲的线程去处理这个任务，做完任务，循环回来等待新任务，等待新任务，
由pthreadPool.h、pthreadPool.cpp两个文件组成。
1、createPthread函数：创建线程全部线程，并将每个线程结构，放入容器中，函数中的goto语句部分，是为了保证所有线程运行起来，并且都处于pthread_cond_wait未激发状态等待。
2、call函数：中pthread_cond_broadcast唤醒一个或者多个线程，并且记录当前工作线程是否够用，每过十秒钟打印一下信息。
3、stopAll函数：唤醒一个或多个线程，并且释放资源
4、inMsgRecvQueueAndSignal函数：将消息插入消息队列中，并调用call函数。
pthreadPool.h
#ifndef __PTHREADPOOL_H_
#define __PTHREADPOOL_H_
#include <vector>
#include <atomic>
#include <pthread.h>
#include <iostream>
#include <list>
#include <unistd.h>
using namespace std;
struct student
{
char name[256];
unsigned int age;
int id;
};
class pthreadPool
{
public:
pthreadPool();
~pthreadPool();
public:
bool createPthread(int threadNUm = 5);
void stopAll();
void call();
void inMsgRecvQueueAndSignal(char* buf);
private:
static void* threadFunc(void* threadData);
void clearMsgRecvQueue();
void msgDispose(char* jobbuf);
private:
struct pthreadItem
{
bool isruning;
pthreadPool* _pThis;
pthread_t _Handle;
pthreadItem(pthreadPool* pthis):_pThis(pthis),isruning(false){}
~pthreadItem(){}
};
private:
static pthread_cond_t m_pthreadCond; // 条件变量
static pthread_mutex_t m_pthreadMutex; // 互斥量
static bool m_shutdown; // 线程退出标识
int m_iThreadNum; // 要创建的线程数
time_t m_iLastTime; // 上次线程不够用，时间记录
atomic<int> m_iRunThreadNum; // 正在运行线程数量 原子操作
vector<pthreadItem*> m_vThread; // 线程容器
list<char*> m_msgRecvQueue; // 消息队列
int m_iRecvQueueCount; // 收消息队列大小
};
#endif // !__PTHREADPOOL_
pthreadPool.cpp文件
#include "pthreadPool.h"
pthread_cond_t pthreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t pthreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
bool pthreadPool::m_shutdown = false;
pthreadPool::pthreadPool()
{
// 运行的线程数为0，时间为0，消息数0
m_iRunThreadNum = 0;
m_iLastTime = 0;
m_iRecvQueueCount = 0;
}
pthreadPool::~pthreadPool()
{
clearMsgRecvQueue();
}
bool pthreadPool::createPthread(int threadNum)
{
if(!threadNum) return false;
m_iThreadNum = threadNum;
pthreadItem* item;
int errCode = 0;
for(int i=0;i<threadNum;i++){
item = new pthreadItem(this);
errCode = pthread_create(&(item->_Handle),NULL,pthreadPool::threadFunc,item);
if(errCode!=0){
cout<<"线程创建失败:"<<i<<endl;
return false;
}
m_vThread.push_back(item);
}
vector<pthreadItem*>::iterator iter;
lblfor:
// goto语句作用，为了所有线程都正常启动，并且卡在pthread_cond_wait()这
for(iter = m_vThread.begin();iter!=m_vThread.end();iter++){
if((*iter)->isruning == false){
usleep(100*1000); // 单位是微秒，100毫秒
goto lblfor;
}
}
return true;
}
void* pthreadPool::threadFunc(void* threadData)
{
pthreadItem* pThread = (pthreadItem*)threadData;
pthreadPool* pPoll = pThread->_pThis;
int errCode = 0;
pthread_t tid = pthread_self();
while(true){
// 拿锁
errCode = pthread_mutex_lock(&m_pthreadMutex);
if(errCode!=0){
cout<<"pthread_mutex_lock fail threadFunc errCode"<<errCode<<endl;
return (void*)0;
}
while((pPoll->m_msgRecvQueue.size() == 0) && m_shutdown == false){
if(pThread->isruning == false)
pThread->isruning = true;
// 整个程序初始化的时候，保证所有线程都卡在这里
// 当线程走到这里，就会释放锁，处于未激发状态
// 一旦被激发，就会去拿锁
pthread_cond_wait(&m_pthreadCond,&m_pthreadMutex);
}
// 判断线程退出条件
if(m_shutdown){
pthread_mutex_unlock(&m_pthreadMutex);
break;
}
// 走到这里可以去消息处理
// 返回第一个元素，没有检查是否存在，走下来就说明有消息
char* jobbuf = pPoll->m_msgRecvQueue.front();
pPoll->m_msgRecvQueue.pop_front();
// 消息队列数减1
--pPoll->m_iRecvQueueCount;
// 可以解锁了
pthread_mutex_unlock(&m_pthreadMutex);
// 能走到这里表示有消息，并且线程正在处理这个消息
// 正在工作的线程数加1，原子操作
++pPoll->m_iRunThreadNum;
// 消息处理
//cout<<"消息处理开始："<<tid<<endl;
//sleep(3);
//cout<<"消息处理结束："<<tid<<endl;
// 消息处理函数
pPoll->msgDispose(jobbuf);
// 消息处理结束
//释放消息内存，运行线程数--
++pPoll->m_iRunThreadNum;
}
return (void*)0;
}
void pthreadPool::msgDispose(char* jobbuf)
{
pthread_t tid = pthread_self();
struct student* stu = (struct student*)jobbuf;
cout<<"tid："<<tid<<" name："<<stu->name<<" age："<<stu->age
<<" id："<<stu->id<<endl;
if(stu!=NULL){
delete stu;
stu = NULL;
}
}
void pthreadPool::call()
{
// 唤醒一个等待该条件的线程，也可能是多个，也就是可以唤醒卡在pthread_cond_wait
int errCode = pthread_cond_signal(&m_pthreadCond);
if(errCode!=0){
cout<<"call fail"<<endl;
return;
}
// 如果工作线程数==开辟线程数需要扩容
if(m_iRunThreadNum == m_iThreadNum)
{
time_t currentime = time(NULL);
if(currentime-m_iLastTime >10)
{
m_iLastTime = currentime;
cout<<"Call()发现线程池中当前空闲线程数量为0，需要考虑扩容"<<endl;
}
}
return;
}
void pthreadPool::stopAll()
{
if(m_shutdown)
return;
m_shutdown = true;
int errCode = pthread_cond_broadcast(&m_pthreadCond);
if(errCode!=0){
cout<<"stopAll faile"<<endl;
return;
}
// 等待所有线程结束
vector<pthreadItem*>::iterator iter;
for(iter=m_vThread.begin();iter!=m_vThread.end();iter++){
pthread_join((*iter)->_Handle,NULL);
if((*iter))
delete *iter;
}
m_vThread.clear();
pthread_cond_destroy(&m_pthreadCond);
pthread_mutex_destroy(&m_pthreadMutex);
cout<<"成功返回，线程池中线程全部正常退出"<<endl;
return;
}
void pthreadPool::clearMsgRecvQueue()
{
while(!m_msgRecvQueue.empty())
{
char* buf = m_msgRecvQueue.front();
m_msgRecvQueue.pop_front();
if(buf!=NULL){
delete buf;
buf = NULL;
}
}
}
void pthreadPool::inMsgRecvQueueAndSignal(char* buf)
{
// 先互斥住
int errCode = pthread_mutex_lock(&m_pthreadMutex);
if(errCode!=0){
cout<<"inMsgRecvQueueAndSignal faile lock"<<endl;
}
m_msgRecvQueue.push_back(buf);
++m_iRecvQueueCount;
errCode = pthread_mutex_unlock(&m_pthreadMutex);
if(errCode!=0){
cout<<"inMsgRecvQueueAndSignal faile unlock"<<endl;
}
// 激发线程做事
call();
return;
}
main函数文件测试代码
#include "pthreadPool.h"
int main()
{
pthreadPool* pool = new pthreadPool();
pool->createPthread(6);
for(int i=0;i<1000;i++){
struct student* stu = new student();
sprintf(stu->name,"name-%d",i);
stu->age = i;
stu->id = 1000+i;
pool->inMsgRecvQueueAndSignal((char*)stu);
}
pool->stopAll();
if(pool!=NULL){
delete pool;
pool = NULL;
}
pthread_exit(0);
}
makefile
all:pthreadPool
pthreadPool:pthreadPool.h pthreadPool.cpp pthreadPoolText.cpp
g++ -o pthreadPool pthreadPool.cpp pthreadPoolText.cpp -pthread -std=c++11
6、主线程先退出对子线程影响
观察一下代码：发现主线程退出之后，子线程没有继续打印，也退出了。
造成原因：主线程执行return 之后调用量glibc库里面的exit函数进行清理处理之后，调用系统调用_exit函数进行进程退出，所以并非是主线程退出，导致子线程退出的。
void* func(void* data)
{
while(true){
cout<<"child loops"<<endl;
}
return NULL;
}
int main()
{
pthread_t pid;
int errCode = pthread_create(&pid,NULL,func,NULL);
sleep(1);
cout<<"main exit"<<endl;
return 0;
}
7、return、exit、pthread_exit区别
return返回到调用者
exit 退出当前进程
pthread_exit退出当前线程
8、进程和线程的区别
进程优缺点
进程优点：具有独立的地址空间，隔离性、稳定性比较好，是由操作系统管理，只要系统不出问题，一个进程的错误不会影响到其他进程。
进程缺点：共享资源麻烦
线程优缺点
线程优点：共享进程的资源，创建销毁，切换简单，速度快，占用内存小，CPU利用率高
线程缺点：需要程序员管理，相互影响几率大，一个线程挂掉将导致整个进程挂掉。
总结
一般需要频繁销毁喝创建，要处理大量运算数据，又要很好显示界面及时性比较高的，因为像这些消耗大量CPU，推荐是一个多线程，一般服务器对稳定性比较高的程序推荐使用多进程。
进程之间是如何通信的
可以通过管道pipe，信号量，共享内存，socket套接字，消息队列， 根据信息量大小，进行选择。
线程之间如何通信的
全局变量。或者自定义的消息通信机制，因为线程间共享进程的资源，所以没有像进程中用于数据交换的方式，它通信的主要目的是为了线程同步
多线程同步和互斥有几种方法
线程同步：互斥锁、信号量、信号量
互斥锁：拥有两种状态，lock和unlock，当互斥锁由某个线程持有时，互斥锁状态就变为lock，之后只有该线程有权利打开锁，其他想要获取互斥锁必须都阻塞，直到解锁。
信号量：信号量是一个计数器，用于控制访问有限共享资源数
条件变量：可以让线程满足特定条件才运行，必须搭配互斥锁一起使用。
