Linux C++线程池实例

http://www.cnblogs.com/danxi/p/6636095.html

想做一个多线程服务器测试程序，因此参考了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\n", ((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\n");
   return 0;
}