Linux C++ 200行完成线程池类

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: pthreadTextpthreadText:pthreadText.cppg++ -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，消息数0m_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_waitint 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:pthreadPoolpthreadPool:pthreadPool.h pthreadPool.cpp pthreadPoolText.cppg++ -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，之后只有该线程有权利打开锁，其他想要获取互斥锁必须都阻塞，直到解锁。
信号量：信号量是一个计数器，用于控制访问有限共享资源数
条件变量：可以让线程满足特定条件才运行，必须搭配互斥锁一起使用。