[C++知识库]c++简单实现一个线程池

#include<iostream>
#include <vector>
#include <pthread.h>
#include <atomic>
using namespace std;
class CThreadPool
{
public:
    CThreadPool()
	{
	}
    ~CThreadPool()
	{
	}
public:
    bool Create(int threadNum);                     //创建该线程池中的所有线程
    void StopAll();                                 //使线程池中的所有线程退出
    void Call();                                    //来任务了，调一个线程池中的线程下来干活
private:
    static void* ThreadFunc(void *threadData);      //新线程的线程回调函数    
private:
    struct ThreadItem
    {
        pthread_t       _Handle;                        //线程句柄
        CThreadPool     *_pThis;                        //记录线程池的指针
        bool            ifrunning;                      //标记是否正式启动起来，启动起来后，才允许调用StopAll()来释放
        ThreadItem(CThreadPool *pthis):_pThis(pthis),ifrunning(false)
        {
        }
        ~ThreadItem()
        {
        }
    };
private:
    static pthread_mutex_t     m_pthreadMutex;      //线程同步互斥量/也叫线程同步锁
    static pthread_cond_t      m_pthreadCond;       //线程同步条件变量
    static bool                m_shutdown;          //线程退出标志，false不退出，true退出
    int                        m_iThreadNum;        //要创建的线程数量
    std::atomic<int>           m_iRunningThreadNum; //线程数, 运行中的线程数，原子操作
    std::vector<ThreadItem *>  m_threadVector;      //线程 容器，容器里就是各个线程了
    std::list<char *>          m_MsgRecvQueue;
};

bool CThreadPool::Create(int threadNum)
{
    ThreadItem *pnew;
    m_iThreadNum =  threadNum;
    int ret = 0;
    for(int i = 0; i < m_iThreadNum; i++)
    {
        m_threadVector.push_back(pnew = new ThreadItem(this));
        ret = pthread_create(&pnew->_Handle, NULL, ThreadFunc, pnew);
        if(ret !=0 )
        {
            printf("create error\n");
        }
    }
    std::vector<ThreadItem *>::iterator pos;
lbfor:
    for(pos = m_threadVector.begin(); pos != m_threadVector.end(); pos++)
    {
        if((*pos)->ifrunning == false)
        {
            usleep(100 * 1000);
            goto lbfor;
        }
    }
    return true;
}


void* CThreadPool::ThreadFunc(void* threadData)
{
    ThreadItem *pThread = static_cast<ThreadItem*>(threadData);
    CThreadPool *pThreadPoolObj = pThread->_pThis;
    CMemory *p_memory = CMemory::GetInstance();
    int err;
    pthread_t tid = pthread_self();
    while(true)
    {
        err = pthread_mutex_lock(&m_pthreadMutex);  
        if(err != 0)
        {
        }
        while((pThreadPoolObj->m_MsgRecvQueue.size() == 0) && m_shutdown == false)
        {
            if(pThread->ifrunning == false)
            {
                pThread->ifrunning = true;
            }
            pthread_cond_wait(&m_pthreadCond, &m_pthreadMutex); 
        }
        if(m_shutdown)
        {
            pthread_mutex_unlock(&m_pthreadMutex); 
            break; 
        }

        err = pthread_mutex_unlock(&m_pthreadMutex); 
        if(err != 0)
        {

        }
        ++pThreadPoolObj->m_iRunningThreadNum;
        --pThreadPoolObj->m_iRunningThreadNum;
    }
    return (void*)0;
}


void CThreadPool::StopAll() 
{
    if(m_shutdown == true)
    {
        return;
    }

    m_shutdown = true;

    int err = pthread_cond_broadcast(&m_pthreadCond); 
    if(err != 0)
    {
        return;
    }

    std::vector<ThreadItem*>::iterator iter;
    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        pthread_join((*iter)->_Handle, NULL);
    }

    pthread_mutex_destroy(&m_pthreadMutex);
    pthread_cond_destroy(&m_pthreadCond);


    for(iter = m_threadVector.begin(); iter != m_threadVector.end(); iter++)
    {
        if(*iter)
            delete *iter;
    }
    m_threadVector.clear();
    return;
}

//来任务了，调一个线程池中的线程下来干活
void CThreadPool::Call()
{
    int err = pthread_cond_signal(&m_pthreadCond);
    if(err != 0 )
    {
		
    }
}
//静态成员初始化
pthread_mutex_t CThreadPool::m_pthreadMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t CThreadPool::m_pthreadCond = PTHREAD_COND_INITIALIZER;
bool CThreadPool::m_shutdown = false;
CThreadPool g_threadpool;
int main(void)
{
}