再手写线程池以及性能分析

---

前言

以前写的线程池文章请参考：线程池的简单实现
本次文章是对线程池的再次学习，也是深度学习哈哈哈。
毕竟人都有遗忘性，常回头看看挺好的哈哈。

---

一、为什么要用线程池

1. 某类任务特别耗时，严重影响该线程处理其他任务
2. 在其他线程中异步执行该任务
3. 线程资源的开销与CPU核心之间平衡选择

线程池的作用

复用线程资源；
减少线程创建和销毁的开销；
可异步处理生产者线程的任务；
减少了多个任务（不是一个任务）的执行时间；

线程池的使用场景

以日志为例，在写日志loginfo(“xxx”)，与日志落盘，是两码事，它们两之间应该是异步的。那么异步解耦就是将日志当作一个任务task，将这个任务抛给线程池去处理，由线程池去负责日志落盘。对于应用程序而言，就可以提升落盘的效率。
以nginx为例，一秒几万的请求，速度很快。如果在其中加一个日志，那么qps一下子就掉下来了，因为每请求一次就需要落盘一次，那么整个服务器的性能就下降。我们可以引入一个线程池，把日志这个任务抛给线程池，对于主循环来说，就只抛任务即可，这样就可以大大提升主线程的效率。这就是线程池异步解耦的作用
不仅仅是日志落盘，还有很多地方都可以用线程池，比较耗时的操作如数据库操作，io处理等，都可以用线程池。
线程池有必要将线程与cpu做亲和性吗？ 在注重cpu处理能力的时候，可以做黏合；如果注重的是异步解耦，那么这里更加注重的是任务，没必要将线程和cpu做绑定。

二、线程池的构成以及相关API的实现

首先我们先来思考下，线程池应该有哪些构成。根据构成来让我们进行代码落实。

首先，我们应该理解的是。这应该是一个生产者和消费者模型。而线程池在其中充当消费者。

在真正实现需求功能的时候。我们首先主程序（生产者）肯定会有比较耗时的需求（任务）需要处理，把任务放到任务队列。woker线程从任务队列中取出任务，执行任务。

接下来先来实现下线程池中相关API。然后在通过实验用例来使用它

首先我们先定义先线程池中的相关变量类型

线程池中的相关变量类型

//任务
typedef struct task_t {
    handler_pt func; //任务的执行函数
    void * arg;  //任务上下文
} task_t;

typedef struct task_queue_t {
    uint32_t head;
    uint32_t tail;
    uint32_t count; //队列的大小
    task_t *queue;  //数组可以一次性分配，避免频繁的malloc(任务队列)
} task_queue_t;

struct thread_pool_t {
    pthread_mutex_t mutex;
    pthread_cond_t condition;
    pthread_t *threads;
    task_queue_t task_queue;

    int closed;  //线程池退出的标记，和连接池中的那个bool类型的一样
    int started; // 当前运行的线程数

    int thrd_count;  //线程的数量
    int queue_size;  //队列的最大值
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

线程池的创建

thread_pool_t *
thread_pool_create(int thrd_count, int queue_size) {
    thread_pool_t *pool;

    if (thrd_count <= 0 || queue_size <= 0) {
        return NULL;
    }

    pool = (thread_pool_t*) malloc(sizeof(*pool));
    if (pool == NULL) {
        return NULL;
    }

    pool->thrd_count = 0; //不能确保所有的线程都启动起来，所以谁启动++就行
    pool->queue_size = queue_size;
    pool->task_queue.head = 0;
    pool->task_queue.tail = 0;
    pool->task_queue.count = 0;

    pool->started = pool->closed = 0;

    pool->task_queue.queue = (task_t*)malloc(sizeof(task_t)*queue_size);
    if (pool->task_queue.queue == NULL) {
        // TODO: free pool
        return NULL;
    }

    //数组
    pool->threads = (pthread_t*) malloc(sizeof(pthread_t) * thrd_count);
    if (pool->threads == NULL) {
        // TODO: free pool
        return NULL;
    }

    int i = 0;
    for (; i < thrd_count; i++) {
        if (pthread_create(&(pool->threads[i]), NULL, thread_worker, (void*)pool) != 0) {
            // TODO: free pool
            return NULL;
        }
        pool->thrd_count++;
        pool->started++;
    }
    return pool;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

创建线程的对象。并根据传来的参数来确定存储任务队列的长度以及可以工作的线程数量并进行创建。

任务线程实现

static void *
thread_worker(void *thrd_pool) {
    thread_pool_t *pool = (thread_pool_t*)thrd_pool;
    task_queue_t *que;
    task_t task;
    //不断的取出任务和执行任务
    for (;;) {
        pthread_mutex_lock(&(pool->mutex));
        que = &pool->task_queue;
        /*用while循环的原因(为什么解除锁定之后还需要循环判断下条件)，原因如下:*/
        // 虚假唤醒   linux  pthread_cond_signal(以前的版本可能会唤醒两个线程)，被唤醒后不一定有任务，虚假唤醒
        // linux 可能被信号唤醒
        // 业务逻辑不严谨，被其他线程抢了该任务(假设有两个线程，其中一个是活跃的，另一个是休眠的。假设有任务到来，活跃的线程取了任务，但是也唤醒了
        // 那个休眠的线程，如果此处不是while，那它就会向下执行报错)
        while (que->count == 0 && pool->closed == 0) { //任务为空，且没有退出线程池，此时就不断的休眠
            // pthread_mutex_unlock(&(pool->mutex))
            // 阻塞在 condition
            // ===================================
            // 解除阻塞
            // pthread_mutex_lock(&(pool->mutex));
            pthread_cond_wait(&(pool->condition), &(pool->mutex));
        }
        if (pool->closed == 1) break;
        task = que->queue[que->head];
        que->head = (que->head + 1) % pool->queue_size;
        que->count--;
        pthread_mutex_unlock(&(pool->mutex));
        (*(task.func))(task.arg);  //执行的任务

    pool->started--;  //执行完一个当前工作的线程数就 减一
    pthread_mutex_unlock(&(pool->mutex));
    pthread_exit(NULL);
    return NULL;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

这里只讲说两点，别的都比较简单，不进行讲述。

第一点：
      此处的while (que->count == 0 && pool->closed == 0)
     为什么任务队列为空且线程池未退出时，进行睡眠的判定用的是while循环而不是if啊。
     即为什么解除锁定之后还需要循环判断下条件
     原因如下：
     	 1. 避免虚假唤醒   linux  pthread_cond_signal(以前的版本可能会唤醒两个线程)，被唤醒后不一定有任务，虚假唤醒
     	 2. linux 可能被信号唤醒
     	 3. 业务逻辑不严谨，被其他线程抢了该任务(假设有两个线程，其中一个是活跃的，另一个是休眠的。假设有任务到来，活跃的线程取了任务，但是也唤醒了那个休眠的线程，如果此处不是while，那它就会向下执行报错)


第二点：
     执行任务的回调函数 (*(task.func))(task.arg);  //执行的任务
     请看如下代码实例便可理解：
1
2
3
4
5
6
7
8
9
10
11
12
13

#include 

typedef char (*PTR_TO_ARR)[30];
typedef int (*PTR_TO_FUNC)(int, int);

int max(int a, int b){
    return a>b ? a : b;
}

char str[3][30] = {
    "http://c.biancheng.net",
    "C-Language"
};

int main(){
    PTR_TO_ARR parr = str;
    PTR_TO_FUNC pfunc = max;
    int i;
   
    printf("max: %d\n", (*pfunc)(10, 20));
    for(i=0; i<3; i++){
        printf("str[%d]: %s\n", i, *(parr+i));
    }

    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

获取任务

//发布任务，生产者线程调用它
int
thread_pool_post(thread_pool_t *pool, handler_pt func, void *arg) {
    if (pool == NULL || func == NULL) {
        return -1;
    }

    task_queue_t *task_queue = &(pool->task_queue);

    if (pthread_mutex_lock(&(pool->mutex)) != 0) {
        return -2;
    }

    if (pool->closed) {
        pthread_mutex_unlock(&(pool->mutex));
        return -3;
    }

    if (task_queue->count == pool->queue_size) {
        pthread_mutex_unlock(&(pool->mutex));
        return -4;
    }

    task_queue->queue[task_queue->tail].func = func;
    task_queue->queue[task_queue->tail].arg = arg;
    task_queue->tail = (task_queue->tail + 1) % pool->queue_size;
    task_queue->count++;

    if (pthread_cond_signal(&(pool->condition)) != 0) {
        pthread_mutex_unlock(&(pool->mutex));
        return -5;
    }
    pthread_mutex_unlock(&(pool->mutex));
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

这里的主要逻辑其实就是生产者（主线程）调用它发布任务，存储到任务队列当中去，并且唤醒一个工作线程，取任务队列中一个任务来进行执行。

线程池的销毁

static void 
thread_pool_free(thread_pool_t *pool) {
    if (pool == NULL || pool->started > 0) {
        return;
    }

    if (pool->threads) {
        free(pool->threads);
        pool->threads = NULL;

        pthread_mutex_lock(&(pool->mutex));
        pthread_mutex_destroy(&pool->mutex);
        pthread_cond_destroy(&pool->condition);
    }

    if (pool->task_queue.queue) {
        free(pool->task_queue.queue);
        pool->task_queue.queue = NULL;
    }
    free(pool);
}

int
wait_all_done(thread_pool_t *pool) {
    int i, ret=0;
    for (i=0; i < pool->thrd_count; i++) {
        if (pthread_join(pool->threads[i], NULL) != 0) {
            ret=1;
        }
    }
    return ret;
}


//清空任务
int
thread_pool_destroy(thread_pool_t *pool) {
    if (pool == NULL) {
        return -1;
    }

    if (pthread_mutex_lock(&(pool->mutex)) != 0) {
        return -2;
    }

    if (pool->closed) {
        thread_pool_free(pool);
        return -3;
    }

    pool->closed = 1;

    if (pthread_cond_broadcast(&(pool->condition)) != 0 || 
            pthread_mutex_unlock(&(pool->mutex)) != 0) {
        thread_pool_free(pool);
        return -4;
    }

    wait_all_done(pool);

    thread_pool_free(pool);
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

这里再来大致总结下，线程的整个实现流程；
首先，是有一个线程组,即有多个工作线程。
即在有一个队列，来存储任务（此线程池实现的比较简单，你也可以定义两个队列，一个存储任务的队列，一个存储执行任务的队列）。
当有任务到来，即调用获取任务的函数，此时把任务存储到一个队列中，然后唤醒一个线程，处理队列中的任务。
当全部程序运行结束，退出线程池时，释放分配的空间，销毁线程即可。ok到此结束，实现完成哈哈哈哈。

接下来看一个调用线程池的小程序：

线程池的使用

#include 
#include 
#include 
#include 

#include "thrd_pool.h"

int nums = 0;
int done = 0;

pthread_mutex_t lock;

void do_task(void *arg) {
    usleep(10000);
    pthread_mutex_lock(&lock);
    done++;
    printf("doing %d task\n", done);
    pthread_mutex_unlock(&lock);
}

int main(int argc, char **argv) {
    int threads = 8;
    int queue_size = 256;


    thread_pool_t *pool = thread_pool_create(threads, queue_size);
    if (pool == NULL) {
        printf("thread pool create error!\n");
        return 1;
    }

    while (thread_pool_post(pool, &do_task, NULL) == 0) {
        pthread_mutex_lock(&lock);
        nums++;
        pthread_mutex_unlock(&lock);
    }

    //printf("add %d tasks\n", nums);
    
    wait_all_done(pool);

    printf("did %d tasks\n", done);
    thread_pool_destroy(pool);
    return 0;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

到此完结。