Linux C++线程池实例

想做一个多线程服务器测试程序，因此参考了github的一些实例，然后自己动手写了类似的代码来加深理解。

目前了解的线程池实现有2种思路：

第一种：

主进程创建一定数量的线程，并将其全部挂起，此时线程状态为idle，并将running态计数为0，等到任务可以执行了，就唤醒线程，此时线程状态为running，计数增加，如果计数达到最大线程数，就再创建一组空闲线程，等待新任务，上一组线程执行完退出，如此交替。

第二种：

采用生成者-消费者模式，主进程作为生成者，创建FIFO队列，在任务队列尾部添加任务，线程池作为消费者在队列头部取走任务执行，这之间有人会提到无锁环形队列，在单生成者单消费者的模式下是有效的，但是线程池肯定是多消费者同时去队列取任务，环形队列会造成挂死。

我的实例采用第二种方式实现，在某些应用场景下，允许一定时间内，任务排队的情况，重复利用已有线程会比较合适。

代码比较占篇幅，因此折叠在下面。

 task.h：

#ifndef TASK_H
#define TASK_H

#include <list>
#include <pthread.h>

using std::list;

struct task {
   void (*function) (void *);
   void *arguments;
   int id;
};

struct work_queue {
   work_queue(){
        pthread_mutex_init(&queue_lock, NULL);
        pthread_mutex_init(&queue_read_lock, NULL);
        pthread_cond_init(&queue_read_cond, NULL);
        qlen = 0;
   }

   ~work_queue() {
       queue.clear();
       pthread_mutex_destroy(&queue_read_lock);
       pthread_mutex_destroy(&queue_lock);
       pthread_cond_destroy(&queue_read_cond);
   }

   void push(task *tsk);
   task *pull();
   void post();
   void wait();

private:
   int qlen;
   list< task * > queue;
   pthread_mutex_t queue_lock;
   pthread_mutex_t queue_read_lock;
   pthread_cond_t  queue_read_cond;
};

#endif

 task.cpp

#include "task.h"

void work_queue::push(task *tsk) {
   pthread_mutex_lock(&queue_lock);
   queue.push_back(tsk);
   qlen++;

   pthread_cond_signal(&queue_read_cond);
   pthread_mutex_unlock(&queue_lock);
}

task* work_queue::pull() {
    wait();

    pthread_mutex_lock(&queue_lock);
    task* tsk = NULL;
    if (qlen > 0) {
        tsk = *(queue.begin());
        queue.pop_front();
        qlen--;

        if (qlen > 0)
            pthread_cond_signal(&queue_read_cond);
    }
    pthread_mutex_unlock(&queue_lock);
    return tsk;
}

void work_queue::post() {
    pthread_mutex_lock(&queue_read_lock);
    pthread_cond_broadcast(&queue_read_cond);
    pthread_mutex_unlock(&queue_read_lock);
}

void work_queue::wait() {
    pthread_mutex_lock(&queue_read_lock);
    pthread_cond_wait(&queue_read_cond, &queue_read_lock);
    pthread_mutex_unlock(&queue_read_lock);
}

threadpool.h

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include "task.h"
#include <vector>

using std::vector;

#define safe_delete(p)  if (p) { delete p;  p = NULL; }

struct threadpool {
    threadpool(int size) : pool_size(size)
                         , thread_list(size, pthread_t(0))
                         , queue(NULL)
                         , finish(false)
                         , ready(0) {

        pthread_mutex_init(&pool_lock, NULL);
    }

    ~threadpool() {
        thread_list.clear();
        safe_delete(queue);
        pthread_mutex_destroy(&pool_lock);
    }

    void init();
    void destroy();
    static void* thread_run(void *tp);

    void incr_ready();
    void decr_ready();
    bool close() const;
    
    work_queue *queue;
    
private:
    int pool_size;
    int ready;
    bool finish;
    pthread_mutex_t pool_lock;
    vector <pthread_t> thread_list;
};


#endif

threadpool.cpp

/*
 * threadpool.cpp
 *
 *  Created on: 2017年3月27日
 *      Author: Administrator
 */

#include "threadpool.h"


void* threadpool::thread_run(void *tp) {
    threadpool *pool = (threadpool *) tp;
    pool->incr_ready();

    while(1) {
        task* tsk = pool->queue->pull();
        if (tsk) {
            (tsk->function)(tsk->arguments);
            delete tsk;
            tsk = NULL;
        }

        if (pool->close())
            break;
    }

    pool->decr_ready();

    return NULL;
}

void threadpool::incr_ready() {
    pthread_mutex_lock(&pool_lock);
    ready++;
    pthread_mutex_unlock(&pool_lock);
}

void threadpool::decr_ready() {
    pthread_mutex_lock(&pool_lock);
    ready--;
    pthread_mutex_unlock(&pool_lock);
}

bool threadpool::close() const {
    return finish;
}

void threadpool::init() {
    queue = new work_queue;
    if (!queue) {
        return;
    }

    for(int i; i<pool_size; i++) {
        pthread_create(&thread_list[i], NULL, threadpool::thread_run, (void *)this);
    }

    while(ready != pool_size) {}
}

void threadpool::destroy() {
    finish = true;

    while(ready) {
        if(queue) {
            queue->post();
        }
    }
}

main.cpp

//============================================================================
// Name        : thread_pool.cpp
// Author      : dancy
// Version     :
// Copyright   : Your copyright notice
// Description : Hello World in C++, Ansi-style
//============================================================================

#include <iostream>
#include "threadpool.h"
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

using namespace std;

void job(void *tsk){
    printf("job %-2d working on Thread #%u
", ((task *)tsk)->id, (int)pthread_self());
}

task *make_task(void (*func) (void *), int id) {
    task *tsk = new task;
    if (!tsk)
        return NULL;
    
    tsk->function = func;
    tsk->arguments = (void *)tsk;
    tsk->id = id;
    
    return tsk;
}

int main() {
    threadpool tp(4);
    tp.init();

    for(int i=0; i<40; i++) 
        tp.queue->push(make_task(&job, i+1));
    
    tp.destroy();
    printf("all task has completed
");
    return 0;
}

以上代码需要在linux下编译，mingw下封装的pthread_t，多了一个void *指针，如果要适配还需要自己再封装一次。