[c/c++]简单的线程池(管理)

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>

#define QUEUE_SIZE 8 //环形队列大小
#define GET_POS(x)  ((x+1)%QUEUE_SIZE)
#define GET_SIZE(x,y)  (x-y)

#ifndef __CBASEDATA_H__
#define __CBASEDATA_H__

class CBaseData
{
public:
    CBaseData(const int& data = 0);
    void set(const int& data);
    int get();
    int m_data;
};

CBaseData::CBaseData(const int& data)
{
    m_data = data;
}

void CBaseData::set(const int& data)
{
    m_data = data;
    printf("_set %d\n",m_data);
}

int CBaseData::get()
{
    printf("_get %d\n",m_data);
    return m_data;
}

#endif

#ifndef __CPV_H__
#define __CPV_H__

  

  //P-V模型，使用环形队列保存

class CPV
{
public:
    static CPV* getInstance();
    int put(CBaseData& data);
    int get(CBaseData& data);

protected:
    CPV();
    ~CPV();
    void _clean();

private:
    CBaseData m_pBaseData[QUEUE_SIZE];//将要保存的是实例，而不是实例的指针(这里不太好，但对指针的操作有点问题，暂时如此)
    int nRead;
    int nWrite;
    pthread_mutex_t lock;
    pthread_cond_t noempty;
    pthread_cond_t nofull;
};

CPV::CPV()
{
    nRead = nWrite = 0;
    pthread_mutex_init(&lock, NULL);
    pthread_cond_init(&noempty, NULL);
    pthread_cond_init(&nofull, NULL);
}

CPV* CPV::getInstance()
{
    static CPV inst;
    return &inst;
}

void CPV::_clean()
{
//nothing
}

CPV::~CPV()
{
    _clean();
    pthread_mutex_destroy(&lock);
    pthread_cond_destroy(&noempty);
    pthread_cond_destroy(&nofull);
}

int CPV::put(CBaseData& data)
{
    pthread_mutex_lock(&lock);
    if( (GET_POS(nWrite) == GET_POS(nRead)) && (nWrite > nRead) )
    {
        pthread_cond_wait(&nofull, &lock);
    }
    m_pBaseData[GET_POS(nWrite)] = data;
    printf("nWrite=%d\n",nWrite);
    nWrite++;

    pthread_cond_signal(&noempty);
    pthread_mutex_unlock(&lock);
    return 0;
}

int CPV::get(CBaseData& data)
{
    pthread_mutex_lock(&lock);
    if( (GET_POS(nWrite) == GET_POS(nRead)) && (nWrite == nRead))
    {
        pthread_cond_wait(&noempty, &lock);
    }
    printf("nRead=%d\n",nRead);
    data = m_pBaseData[GET_POS(nRead)];
    printf("nRead=%d\n",nRead);
    nRead++;

    pthread_cond_signal(&nofull);
    pthread_mutex_unlock(&lock);
    return 0;
}

#endif

//test

 //要传递给线程的配置项

struct thread_conf
{
        int max_jobs;//每个线程执行的任务数，达到该阀值后，线程将退出
};
typedef struct thread_conf thread_conf_t;

static int thread_cnt = 2;//最大消费者线程数
static int cur_thread_cnt = 0;//当前线程数

void *thread_P(void *arg)
{
    CPV *cp = CPV::getInstance();
    int i = 100;
    CBaseData pbd;
    while(1)
    {
        printf("put %d\n",i);
        pbd.set(i);
        cp->put(pbd);
        i++;
        sleep(1);
    }
}

void *thread_V(void *arg)
{
    CPV *cv = CPV::getInstance();
    int cnt = 0;
    thread_conf_t *tc = (thread_conf_t *)arg;
    int max = tc->max_jobs;

    pthread_t tid = pthread_self();
    while(max == 0 || cnt<max)//max为0时，认为没有设置阀值
    {
        CBaseData pbd;
        cv->get(pbd);
        printf("<----thread %d, get %d\n",(int)tid,pbd.get());
        cnt++;
    }
    cur_thread_cnt--;
}

//一个简单的线程池管理者
int thread_manager()
{
    thread_conf_t tc;
    tc.max_jobs = 3;

    pthread_t t_p;
    int ret = pthread_create(&t_p,NULL,thread_P, NULL);
    while(1)
    {
        if( cur_thread_cnt < thread_cnt)
        {
            pthread_t t_v;
            pthread_create(&t_v,NULL,thread_V,(void *)&tc);
            cur_thread_cnt++;
        }
sleep(1);//让cpu喘口气
    }
    return 0;
}


int main()
{
    thread_manager();
    return 0;
}