C语言实现简单线程池

1、为什么要使用线程池？

创建线程和销毁线程的花销是比较大的，频繁创建线程和销毁线程，再加上业务工作线程，消耗系统资源的时间，可能导致系统资源不足。另外，过多的线程会占用太多内存。
针对以上问题，有以下解决思路：

1. 用少量的线程——避免内存占用过多
2. 让这部分线程都保持工作，且可以反复执行任务——避免生命周期的损耗

基于以上思路，所以有了线程池的概念。

2、线程池有什么作用？

1、提高效率
2、统一管理资源
3、合理利用 CPU 和内存

3、线程池适合什么场合？

1. 单个任务处理时间短
2. 待处理任务数量大
3. 实际上，在开发中，如果需要创建5个以上的线程，那么就可以使用线程池来管理

4、线程池实现

线程池模式一般分为两种：HS/HA半同步/半异步模式、L/F领导者与跟随者模式。
在这里，以半同步/半异步模式实现简单线程池。

1. 首先，声明结构体thread_pool用来存储线程信息和互斥句柄

#ifndef _WIN32
#define thrd_t pthread_t
#define mtx_t pthread_mutex_t
#define cond_t pthread_cond_t
#else
#define thrd_t uintptr_t
#define mtx_t SRWLOCK
#define cond_t CONDITION_VARIABLE
#endif


struct thread_pool {
  unsigned int shutdown;     // 监听线程池是否终止
  unsigned int max_threads;  // 线程池最大容量
  unsigned int t_free;       // 空闲线程数(非持久线程数 + 实际空闲线程数)
  unsigned int onfull;       // 任务队列满时新增任务处理0: 直接返回 1：等待空队列
  struct task *tasks, *end;  // 任务信息
  thrd_t* threads;           // 线程信息
  mtx_t lock_ready;
  mtx_t lock_empty;
  cond_t task_ready;  // 任务准备完成
  cond_t task_empty;  // 任务队列未满
};

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

2. 声明结构体task 用来存储任务信息

struct task {
  void* (*execute)(int*, void*);// 执行task的方法，第一个参数固定int型指针，线程结束标志
  void* args;					// 执行task的参数
  int keepalive;				// Task是否为常驻线程
  struct task* next;			// 
};
1
2
3
4
5
6

3. 线程池对外接口

int createPool(struct thread_pool** pool, int max_thread, int onfull);
int destroyPool(struct thread_pool* pool);
int addTaskPool(struct thread_pool* pool, void* (*execute)(int*, void*),
                void* args, int keep_alive);
1
2
3
4

4. 参考

#ifndef _WIN32
static void* execute(void* argvs) {
#else
static u_int __stdcall execute(void* argvs) {
#endif
  int keepalive = 0;
  struct task* task = 0;
  struct thread_pool* pool = (struct thread_pool*)argvs;
  while (!keepalive) {
  	// 加锁
    while (pool->tasks == 0 && pool->shutdown == 0) {
		// 等待task_ready，有任务添加至队列或线程池结束
    }

    if (pool->shutdown) {
		// 释放锁
    }
	// 更新任务队列
	// 释放锁

    task->execute(&pool->shutdown, task->args);

    if (task->keepalive) {// 常驻任务
    	// 增加线程空闲数
    	// 发送task_empty，通知主线程可以增加新任务
    }

    free(task);
  }
Error:
#ifndef _WIN32
  pthread_exit(NULL);
#else
  _endthreadex(0);
#endif
  return 0;
}

int destroyPool(struct thread_pool* pool) {
  struct task* end_task = NULL;

  if (pool->shutdown) {
    return 0;
  }

  pool->shutdown = 1; // 更新shutdown标志

  // 等待全部线程结束
  // 释放资源

  free(pool->threads);

  free(pool);

  return 0;
}

int addTaskPool(struct thread_pool* pool, void* (*execute)(int*, void*),
                void* args, int keep_alive) {
  struct task* work_task = NULL;

  if (execute == NULL) {
    return 0;
  }

  if (pool->shutdown) {
    return 2;
  }

  // 加锁

  while (pool->t_free == 0) {
    if (pool->onfull == 0) {
	  // 释放锁
      return 1;
    }
    
	// 等待task_empty，有常驻任务执行完
  }

#ifndef _WIN32
  pthread_mutex_unlock(&pool->lock_empty);
#else
  ReleaseSRWLockExclusive(&pool->lock_empty);
#endif

  work_task = (struct task*)malloc(sizeof(struct task));
  if (NULL == work_task) {
    return -1;
  }

  work_task->execute = execute;
  work_task->args = args;
  work_task->keepalive = keep_alive;
  work_task->next = NULL;
  // 更新任务队列
  // 发送task_ready，通知线程有新任务添加

  if (keep_alive) { // 常驻任务
    pool->t_free--;
  }

  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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

4、使用示例

#include "stdio.h"
#include "thread/pool.h"

void* execute(int* flag, void* args) {
  printf("currentid = %d\n", args); 
  Sleep(1000);
}

void* execute__live(int* flag, void* args) {
  int i = 0;
  while (*flag == 0) {
    Sleep(100);
    i++;
  } 
  printf("listen = %d\n", i);
}

int main(int args, char** argvs) {
  int ret = 0;
  struct thread_pool* pool = 0x00;

  if ((ret = createPool(&pool, 64, 0)) != 0) {
    goto Err;
  }

  for (int i = 0; i < 32; i++) {
    ret = addTaskPool(pool, execute__live, i, 1);
    if (ret) {
      printf("add err = %d\n", ret);
    }
  }

  for (int i = 0; i < 2048; i++) {
    ret = addTaskPool(pool, execute, i, 0);
    if (ret) {
      printf("add err = %d\n", ret);
    }
  }

  ret = destroyPool(pool);
  printf("err = %d\n", ret);
  pool = 0x00;

Err:
  printf("err = %d\n", ret);
}
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