本文共 8080 字,大约阅读时间需要 26 分钟。
http://blog.csdn.net/tennysonsky/article/details/46490099#
线程池基本原理
在传统服务器结构中,常是有一个总的监听线程监听有没有新的用户连接服务器,每当有一个新的用户进入,服务器就开启一个新的线程用户处理这 个用户的数据包。这个线程只服务于这个用户,当用户与服务器端关闭连接以后,服务器端销毁这个线程。(关于并发服务器更多详情,)。
然而频繁地开辟与销毁线程极大地占用了系统的资源,而且在大量用户的情况下,系统为了开辟和销毁线程将浪费大量的时间和资源。提供了一个解决外部大量用户与服务器有限资源的矛盾。
和传统的一个用户对应一个线程的处理方法不同,它的基本思想就是在程序开始时就在内存中开辟一些线程,线程的数目是固定的,他们独自形成一个类,屏蔽了对外的操作,而服务器只需要将数据包交给线程池就可以了。当有新的客户请求到达时,不是新创建一个线程为其服务,而是从“池子”中选择一个空闲的线程为新的客户请求服务,服务完毕后,线程进入空闲线程池中。如果没有线程空闲的话,就将数据包暂时积累, 等待线程池内有线程空闲以后再进行处理。通过对多个任务重用已经存在的线程对象,降低了对线程对象创建和销毁的开销。当客户请求 时,线程对象已经存在,可以提高请求的响应时间,从而整体地提高了系统服务的表现。
线程池应用实例
一般来说实现一个线程池主要包括以下几个组成部分:
1)线程管理器:用于创建并管理线程池。
2)工作线程:线程池中实际执行任务的线程。在初始化线程时会预先创建好固定数目的线程在池中,这些初始化的线程一般处于空闲状态,一般不占用 CPU,占用较小的内存空间。
3)任务接口:每个任务必须实现的接口,当线程池的任务队列中有可执行任务时,被空闲的工作线程调去执行(线程的闲与忙是通过互斥量实现的),把任务抽象出来形成接口,可以做到线程池与具体的任务无关。
4)任务队列:用来存放没有处理的任务,提供一种缓冲机制,实现这种结构有好几种方法,常用的是队列,主要运用先进先出原理,另外一种是链表之类的数据结构,可以动态的为它分配内存空间,应用中比较灵活,此教程就是用到的链表。
什么时候需要创建线程池呢?简单的说,如果一个应用需要频繁的创建和销毁线程,而任务执行的时间又非常短,这样线程创建和销毁的带来的开销就不容忽视,这时也是线程池该出场的机会了。如果线程创建和销毁时间相比任务执行时间可以忽略不计,则没有必要使用线程池了。
线程池实现示例代码如下:
thread_pool.h 的示例代码:
- #ifndef __THREAD_POOL_H__
- #define __THREAD_POOL_H__
-
- #include <pthread.h>
-
-
-
-
- typedef void *(*pool_task_f)(void *arg);
-
-
-
-
- typedef struct _task{
- pool_task_f process;
- void *arg;
- struct _task *next;
- }pool_task;
-
-
-
-
- typedef struct
- {
- pthread_t *threadid;
- int threads_limit;
- int destroy_flag;
- pool_task *queue_head;
- int task_in_queue;
- pthread_mutex_t queue_lock;
- pthread_cond_t queue_ready;
- }pool_t;
-
-
-
-
-
-
-
-
- void pool_init(pool_t *pool, int threads_limit);
-
-
-
-
-
-
-
- int pool_uninit(pool_t *pool);
-
-
-
-
-
-
-
-
-
- int pool_add_task(pool_t *pool, pool_task_f process, void *arg);
-
-
- #endif
thread_pool.c 的示例代码: - #include <stdio.h>
- #include <stdlib.h>
- #include <pthread.h>
- #include <assert.h>
-
- #include "thread_pool.h"
-
- static void *pool_thread_server(void *arg);
-
-
-
-
-
-
-
-
- void pool_init(pool_t *pool, int threads_limit)
- {
- pool->threads_limit = threads_limit;
- pool->queue_head = NULL;
- pool->task_in_queue = 0;
- pool->destroy_flag = 0;
-
- pool->threadid = (pthread_t *)calloc(threads_limit, sizeof(pthread_t));
- int i = 0;
-
- pthread_mutex_init(&(pool->queue_lock), NULL);
- pthread_cond_init(&(pool->queue_ready), NULL);
-
- for (i = 0; i < threads_limit; i++){
- pthread_create(&(pool->threadid[i]), NULL, pool_thread_server, pool);
- }
- return;
- }
-
-
-
-
-
-
-
- int pool_uninit(pool_t *pool)
- {
- pool_task *head = NULL;
- int i;
-
- pthread_mutex_lock(&(pool->queue_lock));
- if(pool->destroy_flag)
- return -1;
- pool->destroy_flag = 1;
- pthread_mutex_unlock(&(pool->queue_lock));
-
- pthread_cond_broadcast(&(pool->queue_ready));
-
- for (i = 0; i < pool->threads_limit; i++)
- pthread_join(pool->threadid[i], NULL);
- free(pool->threadid);
-
- pthread_mutex_lock(&(pool->queue_lock));
- while(pool->queue_head != NULL){
- head = pool->queue_head;
- pool->queue_head = pool->queue_head->next;
- free(head);
- }
- pthread_mutex_unlock(&(pool->queue_lock));
-
- pthread_mutex_destroy(&(pool->queue_lock));
- pthread_cond_destroy(&(pool->queue_ready));
- return 0;
- }
-
-
-
-
-
-
-
-
-
- static void enqueue_task(pool_t *pool, pool_task_f process, void *arg)
- {
- pool_task *task = NULL;
- pool_task *member = NULL;
-
- pthread_mutex_lock(&(pool->queue_lock));
-
- if(pool->task_in_queue >= pool->threads_limit){
- printf("task_in_queue > threads_limit!\n");
- pthread_mutex_unlock (&(pool->queue_lock));
- return;
- }
-
- task = (pool_task *)calloc(1, sizeof(pool_task));
- assert(task != NULL);
- task->process = process;
- task->arg = arg;
- task->next = NULL;
- pool->task_in_queue++;
- member = pool->queue_head;
- if(member != NULL){
- while(member->next != NULL)
- member = member->next;
- member->next = task;
- }else{
- pool->queue_head = task;
- }
- printf("\ttasks %d\n", pool->task_in_queue);
-
- pthread_cond_signal (&(pool->queue_ready));
- pthread_mutex_unlock (&(pool->queue_lock));
- }
-
-
-
-
-
-
- static pool_task *dequeue_task(pool_t *pool)
- {
- pool_task *task = NULL;
-
- pthread_mutex_lock(&(pool->queue_lock));
-
- if(pool->destroy_flag){
- pthread_mutex_unlock(&(pool->queue_lock));
- printf("thread 0x%lx will be destroyed\n", pthread_self());
- pthread_exit(NULL);
- }
-
- if(pool->task_in_queue == 0){
- while((pool->task_in_queue == 0) && (!pool->destroy_flag)){
- printf("thread 0x%lx is waitting\n", pthread_self());
-
- pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
- }
- }else{
-
- pool->task_in_queue--;
- task = pool->queue_head;
- pool->queue_head = task->next;
- printf("thread 0x%lx received a task\n", pthread_self());
- }
- pthread_mutex_unlock(&(pool->queue_lock));
- return task;
- }
-
-
-
-
-
-
-
-
-
- int pool_add_task(pool_t *pool, pool_task_f process, void *arg)
- {
- enqueue_task(pool, process, arg);
- return 0;
- }
-
-
-
-
-
-
- static void *pool_thread_server(void *arg)
- {
- pool_t *pool = NULL;
-
- pool = (pool_t *)arg;
- while(1){
- pool_task *task = NULL;
- task = dequeue_task(pool);
-
- if(task != NULL){
- printf ("thread 0x%lx is busy\n", pthread_self());
- task->process(task->arg);
- free(task);
- task = NULL;
- }
- }
-
- pthread_exit(NULL);
- return NULL;
- }
下面是测试代码:
- #include <stdio.h>
- #include <unistd.h>
-
- #include "thread_pool.h"
-
- void *task_test(void *arg)
- {
- printf("\t\tworking on task %d\n", (int)arg);
- sleep(1);
- return NULL;
- }
-
- void thread_pool_demo(void)
- {
- pool_t pool;
- int i = 0;
-
- pool_init(&pool, 2);
- sleep(1);
- for(i = 0; i < 5; i++){
- sleep(1);
- pool_add_task(&pool, task_test, (void *)i);
- }
- sleep(4);
-
- pool_uninit(&pool);
- }
-
- int main (int argc, char *argv[])
- {
- thread_pool_demo();
- return 0;
- }
运行结果如下:
参考资料: