自己实现一个简单的线程池

Posted

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了自己实现一个简单的线程池相关的知识，希望对你有一定的参考价值。

参考技术A

[TOC]

线程池的概念这里不多说，在讲它的原理前，我们先自己想一下，如果我来写，那如何实现一个线程池？

首先要定义一个存放所有线程的集合；
另外，每有一个任务分配给线程池，我们就从线程池中分配一个线程处理它。但当线程池中的线程都在运行状态，没有空闲线程时，我们还需要一个队列来存储提交给线程池的任务。

初始化一个线程池时，要指定这个线程池的大小；另外，我们还需要一个变量来保存已经运行的线程数目。

线程池执行任务
接下来就是线程池的核心方法，每当向线程池提交一个任务时。如果 已经运行的线程<线程池大小，则创建一个线程运行任务，并把这个线程放入线程池；否则将任务放入缓冲队列中。

到这里，一个线程池已经实现的差不多了，我们还有最后一个难点要解决：从任务队列中取出任务，分配给线程池中“空闲”的线程完成。

分配任务给线程的第一种思路
很容易想到一种解决思路：额外开启一个线程，时刻监控线程池的线程空余情况，一旦有线程空余，则马上从任务队列取出任务，交付给空余线程完成。

这种思路理解起来很容易，但仔细思考，实现起来很麻烦（1. 如何检测到线程池中的空闲线程 2. 如何将任务交付给一个.start()运行状态中的空闲线程）。而且使线程池的架构变的更复杂和不优雅。

分配任务给线程的第二种思路
现在我们来讲第二种解决思路：线程池中的所有线程一直都是运行状态的，线程的空闲只是代表此刻它没有在执行任务而已；我们可以让运行中的线程，一旦没有执行任务时，就自己从队列中取任务来执行。

为了达到这种效果，我们要重写run方法，所以要写一个自定义Thread类，然后让线程池都放这个自定义线程类

现在我们来总结一下，这个自定义线程池的整个工作过程：

简单线程池的实现

1. 什么是线程池

线程池是线程的集合，拥有若干个线程，线程池中的线程一般用于执行大量的且相对短暂的任务。如果一个任务执行的时间很长，那么就不适合放在线程池中处理，比如说一个任务的执行时间跟进程的生命周期是一致的，那么这个线程的处理就没有必要放到线程池中调度，用一个普通线程即可。


线程池中线程的个数太少的话会降低系统的并发量，太多的话又会增加系统的开销。一般而言，线程池中线程的个数与线程的类型有关系，线程的类型分为


1.     计算密集型任务；

2.     I/O密集型任务。


计算密集型任务是占用CPU资源的，很少被外界的事件打断，CPU的个数是一定的，所以并发数是一定的，因此线程个数等于CPU的个数时是最高效的。


I/O密集型任务意味着执行期间可能会被I/O中断，也就是说这个线程会被挂起，这时的线程个数应该大于CPU的个数。


线程池的本质是生产者与消费者模型的应用。生产者线程向任务队列中添加任务，一旦队列有任务到来，如果线程池有空闲线程，就唤醒空闲线程来执行任务，如果没有空闲线程，并且线程数没有达到阈值（线程池中线程的最大值），就创建新线程来执行任务。


当任务增加的时候能够动态的增加线程池中线程的数量，直到达到一个阈值。这个阈值就是线程池中线程的最大值。


当任务执行完毕时，能够动态的销毁线程池中的线程池。

 

2.线程池的实现

我在 Ubuntu 系统下用C语言写的程序（传送门：github），这是一个非常简单的线程池实现，代码量约300行，仅仅说明线程池的工作原理，我会文章最后给出扩展线程池的思路，使之成为一个拥有C/S架构，socket通信的线程池。


目前而言，用到的知识点就两个：

1.     pthread_mutex_t：互斥锁；

2.     pthread_cond_t：条件变量；


互斥锁和条件变量要配合使用，我在另一篇博客（点击打开链接）里给出了使用方法，有兴趣的童鞋可以去看一下~


我们的小线程池共有五个文件：

1.             condition.h：把互斥锁和条件变量组合在一起，形成一个条件结构体，condition.h就是这个结构体的声明；

2.             condition.c：与condition.h对应，定义了操作条件结构体的函数；

3.             threadpool.h：包含两个结构体，一个是线程控制块（TCB），另一个是线程池结构体，还有三个函数：初始化线程池、销毁线程池、向线程池中添加任务；

4.             threadpool.c：threadpool.h的实现；

5.             main.c：主函数；


除此之外，我使用autotool编译程序，因此还有两个脚本文件：

1.     makefile.am：定义文件之间的依赖关系；

2.     build.sh：编译脚本；

 

接下来让我们深入代码去理解线程池~

 

Condition.h，为了实现线程同步，普遍的做法是将互斥锁（pthread_mutex_t）和条件变量（pthread_cond_t）配合在一起使用，最好的做法就是让两者组合成一个结构体，pthread_mutex_t 和 pthread_cond_t 同属于 pthread.h 头文件：

/*******************************************************************
*  Copyright(c) 2016 Chen Gonghao
*  All rights reserved.
*
*  chengonghao@yeah.net
******************************************************************/

#ifndef _CONDITION_H_
#define _CONDITION_H_

#include <pthread.h>

/* 将互斥锁和条件变量封装成一个结构体 */
typedef struct condition
{
   pthread_mutex_t pmutex ;
   pthread_cond_t pcond ;
} condition_t ;

/* 初始化结构体 */
int condition_init( condition_t* cond ) ;

/* 拿到结构体中的互斥锁 */
int condition_lock( condition_t* cond ) ;

/* 释放结构体中的互斥锁 */
int condition_unlock( condition_t* cond ) ;

/* 使消费者线程等待在条件变量上 */
int condition_wait( condition_t* cond ) ;

/* 使消费者线程等待在条件变量上，abstime：等待超时 */
int condition_timedwait( condition_t* cond, const struct timespec* abstime ) ;

/* 生产者线程通知等待在条件变量上的消费者线程 */
int condition_signal( condition_t* cond ) ;

/* 生产者线程向等待在条件变量上的消费者线程广播 */
int condition_broadcast( condition_t* cond ) ;

/* 销毁结构体 */
int condition_destroy( condition_t* cond ) ;

#endif


 

Condition.c，有8个针对条件结构体的函数，condition.c就是这8个函数的定义，很简单，都是转而调用pthread.h头文件中的库函数。

/*******************************************************************
*  Copyright(c) 2016 Chen Gonghao
*  All rights reserved.
*
*  chengonghao@yeah.net
******************************************************************/

#include "condition.h"

int condition_init( condition_t* cond ) 
{
   int status ;
   if ( ( status = pthread_mutex_init( &cond->pmutex, NULL ) ) )
   {
      return status ;
   }

   if ( ( status = pthread_cond_init( &cond->pcond, NULL ) ) )
   {
      return status ;
   }

   return 0 ;
}

int condition_lock( condition_t* cond ) 
{
   return pthread_mutex_lock( &cond->pmutex) ;
}

int condition_unlock( condition_t* cond ) 
{
   return pthread_mutex_unlock( &cond->pmutex ) ;
}

int condition_wait( condition_t* cond ) 
{
   return pthread_cond_wait( &cond->pcond, &cond->pmutex ) ;
}

int condition_timedwait( condition_t* cond, const struct timespec* abstime ) 
{
   return pthread_cond_timedwait( &cond->pcond, &cond->pmutex, abstime ) ;
}

int condition_signal( condition_t* cond ) 
{
   return pthread_cond_signal( &cond->pcond ) ;
}

int condition_broadcast( condition_t* cond ) 
{
   return pthread_cond_broadcast( &cond->pcond ) ;
}

int condition_destroy( condition_t* cond ) 
{
   int status ;
   if ( ( status = pthread_mutex_destroy( &cond->pmutex ) ) ) 
   {
      return status ;
   }

   if ( ( status = pthread_cond_destroy( &cond->pcond ) ) ) 
   {
      return status ;
   }

   return 0 ;
}


 

Pthreadpool.h：包含两个结构体，线程控制块（TCB）task_t和线程池结构体threadpool，还声明了三个针对线程池的操作：初始化线程池、销毁线程池、向线程池中添加任务。

/*******************************************************************
*  Copyright(c) 2016 Chen Gonghao
*  All rights reserved.
*
*  chengonghao@yeah.net
******************************************************************/

#ifndef _THREAD_POOL_H_
#define _THREAD_POOL_H_

#include "condition.h"

/* 线程控制块（task control block），以单向链表的形式组织 TCB */
typedef struct task
{
   void *( *run ) ( void* arg ) ; // 线程的执行函数
   void *arg ; // 执行函数的参数
   struct task* next ; // 指向下一个 TCB
} task_t ;

/* 线程池结构体 */
typedef struct threadpool
{
   condition_t ready ; // 条件变量结构体
   task_t* first ; // TCB 链表的头指针
   task_t* last ; // TCB 链表的尾指针
   int counter ; // TCB 的总数量
   int idle ; // 空闲 TCB 的个数
   int max_threads ; // 最大线程数量
   int quit ; // 线程池销毁标志
} threadpool_t ;

/* 初始化线程池 */
void threadpool_init( threadpool_t* pool, int threads ) ;

/* 向线程池中添加任务 */ 
void threadpool_add_task( threadpool_t* pool, void* ( *run )( void* arg ), void* arg ) ;

/* 销毁线程池 */
void threadpool_destroy( threadpool_t* pool ) ;

#endif


 

Threadpool.c，共四个函数，对线程池的操作。

/*******************************************************************
*  Copyright(c) 2016 Chen Gonghao
*  All rights reserved.
*
*  chengonghao@yeah.net
******************************************************************/

#include "threadpool.h"
#include <errno.h>
#include <time.h>

/* 线程入口函数 */
void* thread_runtime( void* arg ) 
{
   struct timespec abstime ;
   int timeout ;
   
   // 拿到线程池对象
   threadpool_t* pool = ( threadpool_t* ) arg ;

   while ( 1 ) 
   {
      timeout = 0 ;

      /************************** 进入临界区 ***********************/

      condition_lock( &pool->ready ) ;
      
      // 空闲线程数加 1
      ++ pool->idle ;

      // 如果线程链表为空，而且线程池处于运行状态，那么线程就该等待任务的到来
      while ( pool->first == NULL && pool->quit == 0 )
      {
         printf( "thread 0x%x is waiting\\n", (int)pthread_self() ) ;
         clock_gettime( CLOCK_REALTIME, &abstime ) ;
         abstime.tv_sec += 2 ;
         int status = condition_timedwait( &pool->ready, &abstime ) ;
         if ( status == ETIMEDOUT )
         {
            printf( "thread 0x%x is wait timed out\\n", (int)pthread_self() ) ;
            timeout = 1 ;
            break ;
         }
      }

      // 如果线程等待超时
      if ( timeout && pool->first == NULL )
      {
         -- pool->counter ; // 那么线程数量减 1
         condition_unlock( &pool->ready ) ; // 释放互斥锁
         break ; // 跳出 while，注意，break 之后，线程入口函数执行完毕，线程将不复存在
      }

      // 线程获得任务
      -- pool->idle ; // 线程池空闲数减 1
      if ( pool->first != NULL )
      {
         task_t* t = pool->first ; // 从链表头部取出 TCB
         pool->first = t->next ; // 指向下一个 TCB

         // 执行任务需要一定时间，所以要先解锁
         // 以便生产者可以往链表中加入任务
         // 以及其他消费者可以等待任务
         condition_unlock( &pool->ready ) ;

         t->run( t->arg ) ; // 执行任务的回调函数

         free( t ) ; // 任务执行完毕，销毁 TCB

         condition_lock( &pool->ready ) ;
      } 

      // quit == 1 说明要销毁线程池
      if ( pool->quit && pool->first == NULL )
      {
         -- pool->counter ;
         if ( pool->counter == 0 ) 
         {
            condition_signal( &pool->ready ) ; // 唤醒等待在条件变量上的主线程
         }
         condition_unlock( &pool->ready ) ;
         break ;
      }

      condition_unlock( &pool->ready ) ;

      /************************** 退出临界区 ***********************/
   }

   return NULL ;
}

/* 初始化线程池 */
void threadpool_init( threadpool_t* pool, int threads ) 
{
   condition_init( &pool->ready ) ; // 初始化条件变量结构体
   pool->first = NULL ; // 设置线程链表头指针
   pool->last = NULL ; // 设置线程链表尾指针
   pool->counter = 0 ; // 设置线程池当前线程数
   pool->idle = 0 ; // 设置线程池当前空闲线程数
   pool->max_threads = threads ; // 设置线程池最大线程数
   pool->quit = 0 ; // quit = 0，线程池运行状态；quit = 1，线程池销毁状态
}

/* 向线程池中添加线程 */
void threadpool_add_task( threadpool_t* pool, void* (*run)( void* arg ), void* arg ) 
{
   task_t* newtask = ( task_t* ) malloc ( sizeof( task_t ) ) ; // 创建线程控制块
   newtask->run = run ; // 设置线程的回调函数
   newtask->arg = arg ; // 设置回调函数的参数
   newtask->next = NULL ; // 新加入的线程会被添加到链表尾部

   /************************** 进入临界区 ***********************/

   condition_lock( &pool->ready ) ; // 拿到互斥锁

   // 把新创建的 TCB 添加到线程链表中
   if ( pool->first == NULL ) 
   {
   	  // 如果线程链表为空，则 TCB 作为链表头部
      pool->first = newtask ;
   }
   else
   {
   	  // 如果线程链表不为空，加入到链表尾部
      pool->last->next = newtask ;
   }
   pool->last = newtask ; // 修改链表尾指针
   
   // 如果有空闲线程，那么就唤醒空闲线程
   if ( pool->idle > 0 ) 
   {
      condition_signal( &pool->ready ) ; // 通知等待在条件变量上的空闲线程
   }
   else if ( pool->counter < pool->max_threads ) 
   {
   	  // 如果没有空闲线程可用，而且当前线程数量小于线程池的容量，我们就创建一个线程
      pthread_t tid ;
      pthread_create( &tid, NULL, thread_runtime, pool ) ; // 指定新线程的起始函数为 thread_runtime，把线程池传递给 thread_runtime
      ++ pool->counter ;
   }

   condition_unlock( &pool->ready ) ; // 释放互斥锁

   /************************** 退出临界区 ***********************/
}

/* 销毁线程池 */
void threadpool_destroy( threadpool_t* pool ) 
{
   if ( pool->quit ) 
   {
      return ;
   } 

   /************************** 进入临界区 ***********************/

   condition_lock( &pool->ready ) ;

   // 设置退出标志为真
   pool->quit = 1 ;

   // 如果线程池中正在运行着线程，那么我们需要等待线程执行完毕再销毁
   if ( pool->counter > 0 ) 
   {
      if ( pool->idle > 0 )
      {
         condition_broadcast( &pool->ready ) ;
      }
      while ( pool->counter > 0 ) 
      {
         condition_wait( &pool->ready ) ; // 主线程（main 函数所在线程）将等待在条件变量上
      }
   }

   condition_unlock( &pool->ready ) ;

   /************************** 退出临界区 ***********************/
   
   // 销毁条件变量
   condition_destroy( &pool->ready ) ;
}


 

最后是主函数啦，main.c

/*******************************************************************
*  Copyright(c) 2016 Chen Gonghao
*  All rights reserved.
*
*  chengonghao@yeah.net
******************************************************************/

#include "threadpool.h"

/* 定义线程池最大线程数 */
#define MAX_POOL_SIZE	3

/* 每个任务的回调函数 */
void* mytask( void* arg ) 
{
   printf( " thread 0x%x is working on task %d\\n", (int)pthread_self(), *(int*)arg ) ;
   sleep( 1 ) ;
   free( arg ) ;
   return NULL ;
}

int main( void ) 
{
   threadpool_t pool ; // 定义一个线程池变量
   threadpool_init( &pool, MAX_POOL_SIZE ) ; // 初始化线程池

   // 向线程池中添加 10 个任务，每个任务的处理函数都是 mytask
   for ( int i = 0; i < 10; ++ i ) 
   {
      int* arg = (int*)malloc( sizeof( int ) ) ;
      *arg = i ;
      threadpool_add_task( &pool, mytask, arg ) ;
   }

   threadpool_destroy( &pool ) ; // 销毁线程池
   
   return 0 ;
}


 

画个图来理解线程池的原理：

可以发现生产者的代码主要位于 thread_add_task()函数，消费者的代码主要位于 thread_runtime() 函数。

细心的童鞋也许发现了：thread_runtime()中只有while循环，三个消费者同时执行while 循环，不可能保证循环有序的从生产线上取任务，任务来临时如何保证三个消费者不发生争抢行为呢？这就是条件变量的作用了，当消费者发现生产线为空时，就依次睡在条件变量上，睡在条件变量上的关键代码如下：

int status = condition_timedwait( &pool->ready, &abstime ) ;


如果睡超时了就要自动销毁喔~

当任务来临时，生产者有义务通知睡眠的消费者，唤醒后者，起来嗨~，关键代码如下：

condition_signal(&pool->ready ) ; // 唤醒等待在条件变量上的主线程


执行结果如下：


我设置的线程池容量为 2，任务数是 10，因此有 2 个线程处理 10 个任务。

从上图可以发现，两个线程分别是0xaa1de700 和 0xaa9df700，这两个线程依次处理 0 ~ 9 号任务，没有任务处理时，便进入等待状态（is waiting），等待超时的结果就是自动销毁，然后主线程退出，程序结束。

 

3.扩展线程池

以上就是一个最最小的线程池实现了，这只是一个玩具，仅供了解线程池的实现。但是我们可以把小玩具扩展成一个有模有样的线程池：C/S架构、socket 通信，以下是扩展思路，所有扩展的代码都在：simple_thread_pool/extension/。

客户端：

1.     把 socket 封个包，也就是把原生 socket 封装成一个类，我已经做了这一步，代码在：simple_thread_pool/extension/ossSocket/；

2.     做一个命令工厂类（CommandFactory），根据客户端的输入，生产出一系列的命令对象；

3.     客户端类（Client），其中包含 socket 类和 CommandFactory；

 


服务器端：

       服务器端基本上就是对本文介绍的小线程池的扩展。

1．在main函数中添加监听函数：TCPListener()；

2．TCPListener()的代码主要由 threadpool_add_task() 和 threadpool_runtime()的代码构成；

 

很多基础软件，比如数据库、服务器，底层都跑着一个线程池，要是有时间的话，我会扩展这个小线程池，然后写博客介绍技术实现。

以上是关于自己实现一个简单的线程池的主要内容，如果未能解决你的问题，请参考以下文章

一个简单的线程池实现

初学线程池--1，自己实现一个线程池

简单实现java线程池

多线程使用线程池实现一个简单线程池

一个简单线程池的实现

简单线程池的实现