线程池

1.1, 第一版

在前面我们谈到过, 进程池设计和线程池设计的一些适用场景。

进程池设计:

• 每个进程有独立的内存空间, 增加进程间的隔离性。一个进程崩溃不会影响到其它进程。

• 进程间存在隔离性, 这种解耦促使业务逻辑方便书写。

• 进程的创建和销毁比线程开销大, 占用的内存空间也比线程大。

• 上下文的调度切换时间长。

• 适合并发量低(IO请求数少), 业务复杂(CPU密集), 任务执行事件长的系统设计。

线程池设计:

• 线程之间共享资源, 隔离新差, 一个线程极容易影响到另一个线程(数据同步和一致性)。

• 但是隔离性差, 使得线程间通信比进程间通信要更方便。

• 线程较轻量，创建和销毁的开销较小。

• 适合并发量高(I/O密集型),内存使用要求高, 业务简单, 可以大量快速、轻量级任务处理的场景。

那么当我们真正在设计线程池的时候又该怎么实现那? 其实线程池的设计本质逻辑上和进程池并无太多异常, 只不过我们在线程间通信的是时候, 要远比进程池要简单的多.

1.1.1 设计逻辑

1.1.2 CODE

head

head.h

xxxxxxxxxx
34
1
#ifndef __THREADPOOL__
2
#define __THREADPOOL__
3

4
#include <55header.h>
5
#include "queue.h"
6

7
// 定义线程池
8
typedef struct pool_s{
9
   // 所有子线程id
10
   pthread_t *threadIds;
11
   // 子线程的数量
12
   int threadNum;
13
   // 任务队列
14
   queue_t queue;
15
   // 锁
16
   pthread_mutex_t pool_lock;
17
   // 条件遍历
18
   pthread_cond_t cond;
19
}pool_t;
20

21
// 根据指定数量创建线程池
22
int initPool(pool_t *pPool, int num);
23
// 定义线程的入口函数
24
void *threadMain(void *p);
25

26
// 初始化连接
27
int initTcpSocket(int *socketfd, char *ip, char *port);
28

29
// 添加epoll监听
30
int epoll_addfd(int epollfd, int filefd);
31

32
// 传送文件
33
int sendFile(int net_fd);
34
#endif

mian

mian.c

xxxxxxxxxx
34
1
#include "head.h"
2

3
int main(int argc,char*argv[])
4
{
5
   pool_t pool;
6
   initPool(&pool, 3);
7

8
   int socket_fd;
9
   initTcpSocket(&socket_fd, "192.168.106.129", "8080");
10

11
   int epoll_fd = epoll_create(1);
12
   epoll_addfd(epoll_fd, socket_fd);
13

14
   while(1){
15

16
       struct epoll_event list[1024];
17
       int epoll_num = epoll_wait(epoll_fd, list, 1024, -1);
18

19
       for(int i=0; i<epoll_num; i++){
20

21
           if(list[i].data.fd == socket_fd){
22
               // 有连接进来
23
               int net_fd = accept(socket_fd, NULL, NULL);
24

25
               pthread_mutex_lock( &pool.pool_lock);
26
               enQueue(&pool.queue, net_fd);
27
               pthread_cond_signal(&pool.cond);
28
               pthread_mutex_unlock(&pool.pool_lock);
29
           }
30
       }
31
   }
32

33
   return 0;
34
}

pool

pool.c

xxxxxxxxxx
21
1
#include "head.h"
2

3
int initPool(pool_t *pPool, int num){
4
   // 给子线程id开辟空间
5
   pPool->threadIds = (pthread_t *)calloc(num, sizeof(pthread_t)); 
6
   // 创建线程
7
   for(int i=0; i<num; i++){
8
       pthread_create(&pPool->threadIds[i], NULL, threadMain, pPool);
9
   }
10

11
   // 记录子线程个数
12
   pPool->threadNum = num;
13
   // 初始化任务队列
14
   bzero(&pPool->queue, sizeof(queue_t));
15
   // 初始化锁
16
   pthread_mutex_init(&pPool->pool_lock, NULL);
17
   // 初始化条件变量
18
   pthread_cond_init(&pPool->cond, NULL);
19

20
   return 0;
21
}

tcpInit

tcpInit.c

xxxxxxxxxx
17
1
#include "head.h"
2

3
int initTcpSocket(int * socketfd, char *ip, char *port){
4
   *socketfd = socket(AF_INET, SOCK_STREAM, 0);
5

6
   int reuse = 1;
7
   setsockopt(*socketfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
8

9
   struct sockaddr_in addr;
10
   addr.sin_family = AF_INET;
11
   addr.sin_addr.s_addr = inet_addr(ip);
12
   addr.sin_port = htons(atoi(port));
13

14
   bind(*socketfd, (struct sockaddr *) &addr, sizeof(addr));
15
   listen(*socketfd, 10);
16
   return 0;
17
}

epoll

epoll.c

xxxxxxxxxx
10
1
#include "head.h"
2

3
int epoll_addfd(int epollfd,  int filefd){
4
   struct epoll_event event;
5
   event.data.fd = filefd;
6
   event.events = EPOLLIN;
7
   epoll_ctl(epollfd,EPOLL_CTL_ADD, filefd, &event);
8

9
   return 0;
10
}

worker

worker.c

xxxxxxxxxx
23
1
#include "head.h"
2

3
void *threadMain(void *p){
4
   pool_t *pPool = (pool_t *)p;
5

6
   while(1){
7

8
       pthread_mutex_lock(&pPool->pool_lock );
9
       int net_fd;
10

11
       while(pPool->queue.size <= 0){
12
           pthread_cond_wait(&pPool->cond, &pPool->pool_lock);
13
       }
14
       net_fd = pPool->queue.head->net_fd;
15

16
       deQueue(&pPool->queue);
17
       pthread_mutex_unlock(&pPool->pool_lock);
18

19
       sendFile(net_fd);
20
       close(net_fd);
21
   }
22
   return NULL;
23
}

transFile

transFile.c

xxxxxxxxxx
32
1
#include <sys/sendfile.h>
2

3
typedef struct train_s {
4
   int len;
5
   char buf[1000];
6
} train_t;
7

8
int sendFile(int netFd){
9
   char *file_name = "file.txt";
10
   // 需要O_RDWR,避免mmap权限不足
11
   int file_fd = open(file_name, O_RDWR);
12
   train_t train;
13
   bzero(&train, 0);
14

15
   // 获得文件信息
16
   struct stat stat_file;
17
   fstat(file_fd, &stat_file);
18
   // 发送文件长度
19
   send(netFd, &stat_file.st_size, sizeof(off_t), MSG_NOSIGNAL);
20

21
   // 发送文件长度和名字
22
   bzero(&train, 0);
23
   train.len = strlen(file_name);
24
   memcpy(train.buf, file_name, train.len);
25
   send(netFd, &train,sizeof(int)+train.len, MSG_NOSIGNAL);
26

27
   sendfile(netFd, file_fd,NULL, stat_file.st_size);
28
   printf("over \n");
29

30
   close(file_fd);
31
   return 0;
32
}

queue

queue.h

xxxxxxxxxx
21
1
#ifndef __THREADQUEUE__
2
#define __THREADQUEUE__
3

4
#include <55header.h>
5

6
// 定义队列结点
7
typedef struct node_s{
8
   int net_fd;
9
   struct node_s *pNext;
10
}node_t;
11
// 定义队列
12
typedef struct queue_s{
13
   node_t *head;
14
   node_t *end;
15
   int size;
16
}queue_t;
17

18
int enQueue(queue_t *pQueue, int net_fd);
19
int deQueue(queue_t *pQueue);
20

21
#endif

queue.c

xxxxxxxxxx
39
1
#include "queue.h"
2

3
int enQueue(queue_t *pQueue, int net_fd){
4

5
   // 构建新节点
6
   node_t *pNew = (node_t *)calloc(1, sizeof(node_t));
7
   pNew->net_fd = net_fd;
8

9
   if(pQueue->size == 0){
10
       // 队列为空
11
       pQueue->head = pNew;
12
       pQueue->end = pNew; 
13
   }else{
14
       pQueue->end->pNext  = pNew;
15
       pQueue->end = pNew;
16
   }
17

18
   pQueue->size++;
19

20
   return 0;
21
}
22

23
int deQueue(queue_t *pQueue){
24
   if(pQueue->size == 0){
25
       return -1;
26
   }
27

28
   node_t *p = pQueue->head ;
29
   pQueue->head = p->pNext ;
30

31
   if(pQueue->size == 1){
32
       pQueue->end = NULL;
33
   }
34

35
   pQueue->size--;
36
   free(p);
37

38
   return 0;
39
}

1.2 第二版

1.2.1 线程池的退出: 异步拉起同步

我们不适合在多线程中使用信号(不适合), 因为信号是基于进程设计的, 我们无法确定代码在实际运行过程中, 到底进程中的那个线程收到并且执行了这个信号的处理逻辑, 这可能会导致一些异常问题. 总的来说, 信号的设计和线程的设计从某种程度上是不具有良好的相互适配性.

为了解决上面的问题, 我们可以通过异步拉起同步的方式:

xxxxxxxxxx
1
1
// 既然多线程的程序, 不适合处理信号, 那可以让一个专门的进程(只有一个主线程)的进程, 获取信号, 当这个进程获取到信号之后, 再通过进程间通信的方式, 给最终想根据信号做某些操作的多线程进程发送信息(非信号信息).

1.2.2.1 CODE

main

main.c

xxxxxxxxxx
65
1
#include "head.h"
2

3
int pipe_fd[2];
4
void func(int num){
5
   // 信号触发: 写管道, 随便写点什么
6
   write(pipe_fd[1], "1", 1);
7
   return ;
8
}
9
int main(int argc,char*argv[])
10
{
11
   pipe(pipe_fd);
12
   if(fork() != 0){
13
       // 父进程
14
       // 设置对2号信号的监听
15
       signal(2, func);
16
       wait(NULL);
17

18
       close(pipe_fd[0]);
19
       close(pipe_fd[1]);
20
       exit(0);
21
   }
22
   // 子进程
23
   close(pipe_fd[1]);
24

25
   pool_t pool;
26
   initPool(&pool, 3);
27

28
   int socket_fd;
29
   initTcpSocket(&socket_fd, "192.168.106.129", "8080");
30

31
   int epoll_fd = epoll_create(1);
32
   epoll_addfd(epoll_fd, socket_fd);
33

34
   // 监听管道: 是否有信号
35
   epoll_addfd(epoll_fd, pipe_fd[0]);
36

37
   while(1){
38

39
       struct epoll_event list[1024];
40
       int epoll_num = epoll_wait(epoll_fd, list, 1024, -1);
41

42
       for(int i=0; i<epoll_num; i++){
43

44
           if(list[i].data.fd == socket_fd){
45
               // 有连接进来
46
               int net_fd = accept(socket_fd, NULL, NULL);
47

48
               pthread_mutex_lock( &pool.pool_lock);
49
               enQueue(&pool.queue, net_fd);
50
               pthread_cond_signal(&pool.cond);
51
               pthread_mutex_unlock(&pool.pool_lock);
52
           }
53

54
           // 判断是否有信号触发
55
           if(list[i].data.fd == pipe_fd[0]){
56
               printf("信号触发 \n");
57

58
               char buf[60] = {0};
59
               read(pipe_fd[0], buf, sizeof(buf));
60
           }
61
       }
62
   }
63

64
   return 0;
65
}

1.2.2 线程退出: pthread_cancel

当主线程监听到父进程发过来的信号触发信息, 之后, 我们可以让主线从, 通过pthread_cancel函数,让子线程被动退出.

1.2.2.1 CODE

main

mian.c

xxxxxxxxxx
65
1
#include "head.h"
2

3
int pipe_fd[2];
4
void func(int num){
5
write(pipe_fd[1], "1", 1);
6
return ;
7
}
8
int main(int argc,char*argv[])
9
{
10
pipe(pipe_fd);
11
if(fork() != 0){
12
   signal(2, func);
13
   wait(NULL);
14

15
   close(pipe_fd[0]);
16
   close(pipe_fd[1]);
17
   exit(0);
18
}
19
close(pipe_fd[1]);
20

21
pool_t pool;
22
initPool(&pool, 3);
23

24
int socket_fd;
25
initTcpSocket(&socket_fd, "192.168.106.129", "8080");
26

27
int epoll_fd = epoll_create(1);
28
epoll_addfd(epoll_fd, socket_fd);
29
epoll_addfd(epoll_fd, pipe_fd[0]);
30
while(1){
31

32
   struct epoll_event list[1024];
33
   int epoll_num = epoll_wait(epoll_fd, list, 1024, -1);
34

35
   for(int i=0; i<epoll_num; i++){
36

37
       if(list[i].data.fd == socket_fd){
38
           int net_fd = accept(socket_fd, NULL, NULL);
39
           pthread_mutex_lock( &pool.pool_lock);
40
           enQueue(&pool.queue, net_fd);
41
           pthread_cond_signal(&pool.cond);
42
           pthread_mutex_unlock(&pool.pool_lock);
43
       }
44

45
       // 判断是否有信号触发
46
       if(list[i].data.fd == pipe_fd[0]){
47
           printf("信号触发 \n");
48
           char buf[60] = {0};
49
           read(pipe_fd[0], buf, sizeof(buf));
50

51
           // 通过pthread_cancel, 取消所有子线程
52
           for(int i=0; i<pool.threadNum; i++){
53
               pthread_cancel(pool.threadIds[i]);
54
           }
55

56
           for(int i=0; i<pool.threadNum; i++){
57
               pthread_join(pool.threadIds[i], NULL);
58
           }
59
           exit(0);
60
       }
61
   }
62
}
63

64
return 0;
65
}

但是当我们真正在代码执行的时候, (ps -elLf)我们发现好像子线程并没有完全退出

xxxxxxxxxx
2
1
// 原因是因为, 无论pthread_cond_wait是被pthread_cond_signal唤醒或者广播唤醒, 或者被进程要求被动退出唤醒, 只要被唤醒第一件事情就是获取锁
2
// 又因为pthread_cond_wait是取消点函数, 这回导致进程带锁死亡, 导致别的进程醒来无法获取锁, 产生死锁

1.2.2.2 CODE: 改进方案

在上面的问题的基础上, 我们可以优化资源清理行为. 我们知道pthread_cancel函数导致线程退出, 会自动执行被退出线程的清理函数, 所以我们可以在加锁的时候, 增加清理函数, 防止死锁

worker

worker.c

xxxxxxxxxx
31
1
#include "head.h"
2

3
void cleanLock(void *arg){
4
   pool_t *pPool = (pool_t *) arg;
5

6
   pthread_mutex_unlock(&pPool->pool_lock);
7
}
8
void *threadMain(void *p){
9
   pool_t *pPool = (pool_t *)p;
10

11
   while(1){
12

13
       int net_fd;
14
       pthread_mutex_lock(&pPool->pool_lock );
15
       // 调用清理函数
16
       pthread_cleanup_push(cleanLock, pPool);
17

18
       while(pPool->queue.size <= 0){
19
           pthread_cond_wait(&pPool->cond, &pPool->pool_lock);
20
       }
21
       net_fd = pPool->queue.head->net_fd;
22

23
       deQueue(&pPool->queue);
24
       //pthread_mutex_unlock(&pPool->pool_lock);
25
       pthread_cleanup_pop(1);    
26

27
       sendFile(net_fd);
28
       close(net_fd);
29
   }
30
   return NULL;
31
}

1.2.3 更好的退出方式: 标记位退出

虽然我们上面想了一些办法, 可以使线程退出, 但是上面的退出方式本质上是一种被动退出的方式, 某种程度上这不是一个良好的退出方式, 因为子线程没有办法到底在那个代码运行节点退出, 这可能导致有些资源没有正常释放或者某些必要的逻辑进行了一半, 就使得线程退出.

所以在上述逻辑的基础上, 我们可以取消被动退出的方式, 转而使用一种, 以标志位标记的主动退出模式.

1.2.3.1 CODE

head

head.h: 增加退出标记位

xxxxxxxxxx
36
1
#ifndef __THREADPOOL__
2
#define __THREADPOOL__
3

4
#include <55header.h>
5
#include "queue.h"
6

7
// 定义线程池
8
typedef struct pool_s{
9
// 所有子线程id
10
pthread_t *threadIds;
11
// 子线程的数量
12
int threadNum;
13
// 任务队列
14
queue_t queue;
15
// 锁
16
pthread_mutex_t pool_lock;
17
// 条件遍历
18
pthread_cond_t cond;
19
// 退出标记
20
int exitFlag;
21
}pool_t;
22

23
// 根据指定数量创建线程池
24
int initPool(pool_t *pPool, int num);
25
// 定义线程的入口函数
26
void *threadMain(void *p);
27

28
// 初始化连接
29
int initTcpSocket(int *socketfd, char *ip, char *port);
30

31
// 添加epoll监听
32
int epoll_addfd(int epollfd, int filefd);
33

34
// 传送文件
35
int sendFile(int net_fd);
36
#endif

main

main.c

xxxxxxxxxx
66
1

2
int pipe_fd[2];
3
void func(int num){
4
write(pipe_fd[1], "1", 1);
5
return ;
6
}
7
int main(int argc,char*argv[])
8
{
9
pipe(pipe_fd);
10
if(fork() != 0){
11
signal(2, func);
12
wait(NULL);
13

14
close(pipe_fd[0]);
15
close(pipe_fd[1]);
16
exit(0);
17
}
18
setpgid(0, 0);
19
close(pipe_fd[1]);
20

21
pool_t pool;
22
initPool(&pool, 3);
23

24
int socket_fd;
25
initTcpSocket(&socket_fd, "192.168.106.129", "8080");
26

27
int epoll_fd = epoll_create(1);
28
epoll_addfd(epoll_fd, socket_fd);
29
epoll_addfd(epoll_fd, pipe_fd[0]);
30
while(1){
31

32
struct epoll_event list[1024];
33
int epoll_num = epoll_wait(epoll_fd, list, 1024, -1);
34

35
for(int i=0; i<epoll_num; i++){
36

37
  if(list[i].data.fd == socket_fd){
38
      int net_fd = accept(socket_fd, NULL, NULL);
39
      pthread_mutex_lock( &pool.pool_lock);
40
      enQueue(&pool.queue, net_fd);
41
      pthread_cond_signal(&pool.cond);
42
      pthread_mutex_unlock(&pool.pool_lock);
43
  }
44

45
  // 判断是否有信号触发
46
  if(list[i].data.fd == pipe_fd[0]){
47
      printf("信号触发 \n");
48
      char buf[60] = {0};
49
      read(pipe_fd[0], buf, sizeof(buf));
50

51
      // 修改标志位
52
      pthread_mutex_lock(&pool.pool_lock);
53
      pool.exitFlag = 1;
54
      pthread_cond_broadcast(&pool.cond);
55
      pthread_mutex_unlock(&pool.pool_lock);
56

57
      for(int i=0; i<pool.threadNum; i++){
58
          pthread_join(pool.threadIds[i], NULL);
59
      }
60
      exit(0);
61
  }
62
}
63
}
64

65
return 0;
66
}

worker

worker.c

xxxxxxxxxx
28
1
#include "head.h"
2

3
void *threadMain(void *p){
4
pool_t *pPool = (pool_t *)p;
5

6
while(1){
7

8
   int net_fd;
9
   pthread_mutex_lock(&pPool->pool_lock );
10

11
   while(pPool->queue.size <= 0 && pPool->exitFlag ==0 ){
12
       pthread_cond_wait(&pPool->cond, &pPool->pool_lock);
13
   }
14
   if(pPool->exitFlag == 1){
15
       printf("son thread exit \n");
16
       pthread_mutex_unlock(&pPool->pool_lock);
17
       pthread_exit(NULL);
18
   }
19
   net_fd = pPool->queue.head->net_fd;
20

21
   deQueue(&pPool->queue);
22
   pthread_mutex_unlock(&pPool->pool_lock);
23

24
   sendFile(net_fd);
25
   close(net_fd);
26
}
27
return NULL;
28
}