linux線程池分析

一. 線程池學習文件

pool_test/  -> 線程池函數接口實現源碼，簡單實例。

系統編程項目接口設計說明書.doc  -> 詳細說明了線程池各個函數的頭文件/原型/參數/返回值..。

線程池模型.jpg  -> 幫助大家理解線程池原理。

二. 學習線程池實現過程？

1. 什么是線程池？

線程池就是多個線程組合起來的一個集合，當有任務時，線程就會處理任務，當沒有任務時，線程休息。

2. 分析線程池源碼

thread_pool.c  -> 線程池函數接口源碼

thread_pool.h  -> 函數接口聲明/結構體聲明/頭文件..

===========================================================

thread_pool.h

 

#define MAX_WAITING_TASKS  1000

   -> 最大的任務等待個數

#define MAX_ACTIVE_THREADS 20     -> 最大線程個數

 

0）任務節點結構體

struct task

{

      

       void *(*do_task)(void *arg);  -> 任務函數

       void *arg;

 

                    -> 任務函數的參數

       struct task *next;         -> 指向下一個任務節點的指針

};

 

1）線程池模型

typedef struct thread_pool

{

      

       pthread_mutex_t lock;

  -> 互斥鎖

       pthread_cond_t  cond;

  -> 條件變量   

       bool shutdown;

              -> 線程池關閉標識符號  true->關閉   false->未關閉

       struct task *task_list;

-> 任務隊列的頭文件

       pthread_t *tids;

       -> 存放線程TID號空間地址     

       unsigned max_waiting_tasks;

 -> 最大的等待任務的個數     

       unsigned waiting_tasks;  -> 當前等待任務的個數

      

       unsigned active_threads;

  -> 當前線程池中線程的個數

}thread_pool;

2）初始化線程池函數模型

bool init_pool(thread_pool *pool, unsigned int threads_number);

3）線程處理函數

void *routine(void *arg)

4）添加任務函數

bool add_task(thread_pool *pool,void *(*do_task)(void *arg), void *arg)

===========================================================

thread_pool.c

1）初始化線程池函數源碼

 

2）線程處理函數源碼

 

3）添加任務函數源碼

 

源碼：

頭文件：

#ifndef _THREAD_POOL_H_
#define _THREAD_POOL_H_

#include <stdio.h>
#include <stdbool.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>

#include <errno.h>
#include <pthread.h>

#define MAX_WAITING_TASKS    1000
#define MAX_ACTIVE_THREADS    20

struct task
{
    void *(*do_task)(void *arg);
    void *arg;

    struct task *next;
};

typedef struct thread_pool
{
    pthread_mutex_t lock;
    pthread_cond_t  cond;
    bool shutdown;
    struct task *task_list;
    pthread_t *tids;
    unsigned max_waiting_tasks;
    unsigned waiting_tasks;
    unsigned active_threads;
}thread_pool;


bool init_pool(thread_pool *pool, unsigned int threads_number);
bool add_task(thread_pool *pool, void *(*do_task)(void *arg), void *task);
int  add_thread(thread_pool *pool, unsigned int additional_threads_number);
int  remove_thread(thread_pool *pool, unsigned int removing_threads_number);
bool destroy_pool(thread_pool *pool);

void *routine(void *arg);


#endif

 

功能函數：

#include "thread_pool.h"

void handler(void *arg)
{
    printf("[%u] is ended.\n",
        (unsigned)pthread_self());

    //解鎖！
    pthread_mutex_unlock((pthread_mutex_t *)arg);
}

void *routine(void *arg)
{
    //接住線程池的地址
    thread_pool *pool = (thread_pool *)arg;
    struct task *p;

    while(1)
    {
        //取消例程函數，將來線程上鎖了，如果收到取消請求，那么先解鎖，再退出
        pthread_cleanup_push(handler, (void *)&pool->lock);
        
        //任務隊列是屬於臨界資源。
        //訪問任務隊列之前都必須上鎖。
        pthread_mutex_lock(&pool->lock);
    
        //如果當前線程池未被關閉並且線程池中沒有需要處理的任務時：
        while(pool->waiting_tasks == 0 && !pool->shutdown)
        {
            //那么就進入條件變量中等待！
            pthread_cond_wait(&pool->cond, &pool->lock);
        }
        
        //如果線程等待任務為0，並且線程池已經關閉了。
        if(pool->waiting_tasks == 0 && pool->shutdown == true)
        {    
            //解鎖
            pthread_mutex_unlock(&pool->lock);    
            
            //走人
            pthread_exit(NULL); 
        }

        //有任務做，代表肯定不是空鏈表，拿任務p
        p = pool->task_list->next;
        pool->task_list->next = p->next;
        
        //當前等待的任務的個數-1
        pool->waiting_tasks--;

        //解鎖
        pthread_mutex_unlock(&pool->lock);
        
        //刪除線程取消例程函數
        pthread_cleanup_pop(0);
        
        //設置線程不可以響應取消。
        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
        
        //執行任務節點中函數過程中，不希望被別人取消掉。
        (p->do_task)(p->arg);
        
        //設置為可以響應取消
        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);

        //釋放任務節點p的內存空間
        free(p);
    }

    pthread_exit(NULL);
}

bool init_pool(thread_pool *pool, unsigned int threads_number)
{
    //1. 初始化互斥鎖
    pthread_mutex_init(&pool->lock, NULL);
    
    //2. 初始化條件變量
    pthread_cond_init(&pool->cond, NULL);

    //3. 線程池關閉標志為未關閉
    pool->shutdown = false;
    
    //4. 為任務隊列頭節點申請空間
    pool->task_list = malloc(sizeof(struct task));
    
    //5. 為線程TID號申請空間
    pool->tids = malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS);

    //錯誤判斷
    if(pool->task_list == NULL || pool->tids == NULL)
    {
        perror("allocate memory error");
        return false;
    }

    //為任務隊列頭節點的指針域賦值NULL
    pool->task_list->next = NULL;
    
    //設置最大等待任務個數為1000
    pool->max_waiting_tasks = MAX_WAITING_TASKS;
    
    //設置當前等待任務的個數為0
    pool->waiting_tasks = 0;
    
    //設置當前線程池線程的個數
    pool->active_threads = threads_number;

    int i;
    //創建線程池中的子線程
    for(i=0; i<pool->active_threads; i++)
    {
        if(pthread_create(&((pool->tids)[i]), NULL,routine, (void *)pool) != 0)
        {
            perror("create threads error");
            return false;
        }
    }

    //初始化成功
    return true;
}

bool add_task(thread_pool *pool,void *(*do_task)(void *arg), void *arg)
{
    //為新節點申請內存空間
    struct task *new_task = malloc(sizeof(struct task));
    if(new_task == NULL)
    {
        perror("allocate memory error");
        return false;
    }
    
    //為新節點的數據域賦值
    new_task->do_task = do_task;  //函數
    new_task->arg = arg; //函數的參數
    
    //為新節點的指針域賦值
    new_task->next = NULL;  

    //訪問任務隊列前，先上鎖！
    pthread_mutex_lock(&pool->lock);
    
    //如果當前等待任務個數>=1000，則添加任務失敗！
    if(pool->waiting_tasks >= MAX_WAITING_TASKS)
    {
        //解鎖
        pthread_mutex_unlock(&pool->lock);

        //輸出錯誤信息
        fprintf(stderr, "too many tasks.\n");
        
        //釋放剛剛初始化過的新節點
        free(new_task);

        return false;
    }
    
    //尋找任務隊列的最后一個節點
    struct task *tmp = pool->task_list;
    while(tmp->next != NULL)
        tmp = tmp->next;
    
    //tmp->next = NULL;

    //把新節點尾插進去任務隊列中
    tmp->next = new_task;
    
    //當前最大的等待任務的個數+1
    pool->waiting_tasks++;

    //解鎖
    pthread_mutex_unlock(&pool->lock);

    //隨機喚醒條件變量中其中一個線程起來就可以了。
    pthread_cond_signal(&pool->cond);
    
    return true;
}

int add_thread(thread_pool *pool, unsigned additional_threads)
{
    //如果新增0個線程
    if(additional_threads == 0)
        return 0; //直接返回0

    //添加后線程總數
    unsigned total_threads = 
            pool->active_threads + additional_threads;
                        
    int i, actual_increment = 0;
    
    //創建線程
    for(i = pool->active_threads;  
          i < total_threads && i < MAX_ACTIVE_THREADS;  
          i++) 
       {
        if(pthread_create(&((pool->tids)[i]),NULL, routine, (void *)pool) != 0)
        {
            perror("add threads error");
            if(actual_increment == 0) 
                return -1;

            break;
        }
        actual_increment++;  //真正創建的線程個數
    }

    //當前活躍的線程數 = 原來活躍的線程數 + 新實際創建的線程數
    pool->active_threads += actual_increment;
    
    return actual_increment;
}

     
int remove_thread(thread_pool *pool, unsigned int removing_threads)
{
    //如果需要刪除0條線程
    if(removing_threads == 0)
        return pool->active_threads; //當前線程池活躍的線程個數

    //剩余的線程數 = 當前活躍的線程數 - 需要刪除的線程。
    int remaining_threads = pool->active_threads - removing_threads;

    //線程池中至少有1條線程
    remaining_threads = remaining_threads > 0 ? remaining_threads : 1;

    int i;  
    for(i=pool->active_threads-1; i>remaining_threads-1; i--)
    {    
        errno = pthread_cancel(pool->tids[i]);
        if(errno != 0)
            break;
    }

    //如果取消失敗，則函數返回-1
    if(i == pool->active_threads-1)
        return -1;
    else
    {
        //計算當前剩余實際的個數
        pool->active_threads = i+1;
        return i+1; //返回當前線程剩余的個數
    }
}


bool destroy_pool(thread_pool *pool)
{
    pool->shutdown = true; //當前線程池標志是關閉狀態
    pthread_cond_broadcast(&pool->cond);
    int i;
    for(i=0; i<pool->active_threads; i++)
    {
        errno = pthread_join(pool->tids[i], NULL);

        if(errno != 0)
        {
            printf("join tids[%d] error: %s\n",
                    i, strerror(errno));
        }
    
        else
            printf("[%u] is joined\n", (unsigned)pool->tids[i]);
        
    }


    free(pool->task_list);
    free(pool->tids);
    free(pool);

    return true;
}

 

主函數：

#include "thread_pool.h"

void *mytask(void *arg) //線程的任務
{
    int n = (int)arg;

    //工作任務：余數是多少，就睡多少秒，睡完，任務就算完成
    printf("[%u][%s] ==> job will be done in %d sec...\n",
        (unsigned)pthread_self(), __FUNCTION__, n);

    sleep(n);

    printf("[%u][%s] ==> job done!\n",
        (unsigned)pthread_self(), __FUNCTION__);

    return NULL;
}

void *count_time(void *arg)
{
    int i = 0;
    while(1)
    {
        sleep(1);
        printf("sec: %d\n", ++i);
    }
}

int main(void)
{
    // 本線程用來顯示當前流逝的秒數
    // 跟程序邏輯無關
    pthread_t a;
    pthread_create(&a, NULL, count_time, NULL);

    // 1, initialize the pool
    thread_pool *pool = malloc(sizeof(thread_pool));
    init_pool(pool, 2);
    //2個線程都在條件變量中睡眠

    // 2, throw tasks
    printf("throwing 3 tasks...\n");
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));

    // 3, check active threads number
    printf("current thread number: %d\n",
            remove_thread(pool, 0));//2
    sleep(9);

    // 4, throw tasks
    printf("throwing another 6 tasks...\n");
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    add_task(pool, mytask, (void *)(rand()%10));
    
    // 5, add threads
    add_thread(pool, 2);

    sleep(5);

    // 6, remove threads
    printf("remove 3 threads from the pool, "
           "current thread number: %d\n",
            remove_thread(pool, 3));

    // 7, destroy the pool
    destroy_pool(pool);
    return 0;
}