Chinaunix首页 | 论坛 | 博客

苦海澜石

首页　| 　博文目录　| 　关于我

khls27

  • 博客访问： 758256
  • 博文数量： 144
  • 博客积分： 0
  • 博客等级： 民兵
  • 技术积分： 1150
  • 用 户 组： 普通用户
  • 注册时间： 2014-03-17 14:32
个人简介

小公司研发总监，既当司令也当兵！

文章分类

全部博文（144）

推荐博文
相关博文
池化技术与线程池的简单实现

分类： LINUX

2020-08-18 10:13:56

池化技术

在系统开发过程中，我们经常会用到池化技术。通俗的讲，池化技术就是：把一些资源预先分配好，组织到对象池中，之后的业务使用资源从对象池中获取，使用完后放回到对象池中。这样做带来几个明显的好处：

1），资源重复使用， 减少了资源分配和释放过程中的系统消耗。比如，在IO密集型的服务器上，并发处理过程中的子线程或子进程的创建和销毁过程，带来的系统开销将是难以接受的。所以在业务实现上，通常把一些资源预先分配好，如线程池，数据库连接池，Redis连接池，HTTP连接池等，来减少系统消耗，提升系统性能。

2），可以对资源的整体使用做限制。这个好理解，相关资源预分配且只在预分配是生成，后续不再动态添加，从而限制了整个系统对资源的使用上限。类似一个令牌桶的功能。

3），池化技术分配对象池，通常会集中分配，这样有效避免了碎片化的问题。

线程池

有了上面的池化技术铺垫，理解线程池就很容易了：

1）先启动若干数量的线程，并让这些线程都处于睡眠状态，并以一定的方式组织起来，形成“线程池”；

2）当客户端有一个新请求时，就会从线程池中选取一个线程，并唤醒线程，让它来处理客户端的这个请求；

3）当处理完这个请求后，线程又处于睡眠状态，并且放回到线程池中，等待任务到来调度。

线程池设计


 如上图，是基于线程条件变量唤醒的简单线程池模型，主要包括了：任务，任务队列，工作线程几个组件。说明如下：

任务（job）

任务是对一次可执行过的描述，通常包括执行的方法（执行函数）和数据。比如，一个报文接收过程就是一个任务，包括：报文处理函数和报文内容。本例中，对job的定义如下：

点击(此处)折叠或打开

  1. typedef struct thread_job {
  2.     struct list_head list;
  3.     struct thread_pool* pool;
  4.     job_process_func jfunc; // 任务执行函数
  5.     void* data; // 任务数据，由调用者分配，框架释放
  6. } job_t

任务队列（job_queue）

任务队列缓存着所有待执行的任务，供各个工作线程调度。当工作线程空闲或被唤醒时，会尝试从任务队列中摘取一个任务，然后在线程空间内执行。任务队列用线程互斥锁进行保护。为了方便理解整个任务执行转换的过程，本例中，把空闲任务队列单独出来了。 所以，任务的状态有四种： 
    1）在空闲对列中等待业务；
    2）在业务流程中装配；
    3）在工作队列中等待调度；

    4）在工作线程中执行。

工作线程

     工作线程等待在条件变量上（job_dequeue_valid，新任务需要处理），当新任务到达时，线程被唤醒。线程会尝试从任务队列上获取任务，并执行它；执行完毕后重新等待在条件变量上。如此往复：

点击(此处)折叠或打开

  1. while (1)
  2.     {
  3.         if (thrdp_stopping(pool))
  4.         {
  5.             pthread_exit(NULL);
  6.         }

  8.         pthread_mutex_lock(&(pool->qlock));
  9.         while (list_empty(&pool->job_queue))
  10.         {
  11.             // nothing to process, wait until job come in
  12.             printf("thread<%u> get none job, hangup...\n", (unsigned int)pthread_self() );
  13.             pthread_cond_wait(&(pool->job_dequeue_valid), &(pool->qlock));

  15.             if (thrdp_stopping(pool))
  16.             {
  17.                 pthread_exit(NULL);
  18.             }

  20.         }
  21.         job = list_first_entry(&pool->job_queue, struct thread_job, list);
  22.         list_del(&job->list);
  23.         pthread_mutex_unlock(&(pool->qlock));
  24.         job->jfunc(job);
  25.     }

如上代码片段，特别提醒一下，当工作线程从等待条件中被唤醒时，一定要对当前的执行条件再检查，原因是当前线程可能是被“虚假唤醒”。

比如，当前队列是空的，有A，B，C三个工作线程都在等待任务；当一个任务R0到来时，A,B,C三个线程均可能被唤醒；由于任务队列锁的关系，其中只有一个（假如A）取得的实际任务，而B和C均未取得任务，因此B和C称为被“虚假唤醒”。此时，B和C必须对任务获取状态进行再次判断，确保正常获取到任务，方能往下执行调度，否则，应该重新休眠等待任务。


代码实现

附完整代码和测试代码threadpool.h


点击(此处)折叠或打开

  1. #ifndef __THREAD_POOL_H__
  2. #define __THREAD_POOL_H__
  3. #include <linux/types.h>
  4. #include <pthread.h>
  5. #include <syslog.h>
  6. #include <sys/stat.h>
  7. #include <unistd.h>
  8. #include <stdio.h>
  9. #include <stdlib.h>
  10. #include <sys/socket.h>
  11. #include <sys/types.h>
  12. #include <string.h>
  13. #include <asm/types.h>
  14. #include <linux/netlink.h>
  15. #include <linux/socket.h>
  16. #include <linux/types.h>
  17. #include <sys/socket.h>
  18. #include <netinet/in.h>
  19. #include <arpa/inet.h>
  20. #include <errno.h>
  21. #include <unistd.h>

  23. #include "list.h"


  26. #define THRDP_STATUS_STOPPING 0x01 // 线程池是否已经关闭

  28. typedef void (*job_process_func)(void*);

  30. struct thread_pool
  31. {
  32.     int thread_num; // 线程池中开启线程的个数
  33.     int max_job_num; // 队列中最大job的个数
  34.     struct list_head job_queue; // 待处理的job队列
  35.     struct list_head free_job_queue; // job空闲池


  38.     pthread_t *pthreads; //线程池中所有线程的pthread_t
  39.     pthread_mutex_t qlock; //互斥信号量,保护job_queue
  40.     pthread_mutex_t fqlock; //互斥信号量,保护free_job_queue
  41.     pthread_cond_t job_dequeue_valid; // 隊列不為空,需要喚醒線程處理
  42.     pthread_cond_t job_enqueue_valid; // 隊列沒有滿,可以繼續添加job

  44.     int flags;
  45. };

  47. typedef struct thread_job {
  48.     struct list_head list;
  49.     struct thread_pool* pool;
  50.     job_process_func jfunc; // 任务执行函数
  51.     void* data; // 任务数据，由调用者分配，框架释放
  52. } job_t;

  54. struct thread_pool* THRDP_new(unsigned int thread_num, unsigned int max_job_num);
  55. void THRDP_destroy(struct thread_pool* pool);
  56. struct thread_job* THRDP_job_get(struct thread_pool* pool);
  57. void THRDP_job_put(struct thread_pool* pool, struct thread_job* job);
  58. void THRDP_job_add(struct thread_pool* pool, struct thread_job* job);

  60. #endif

    注意，上面引用了list.h，请查看我之前发的用户态list的头文件。
    threadpool.c

点击(此处)折叠或打开

  1. #include "threadpool.h"


  4. static inline int thrdp_stopping (struct thread_pool* pool)
  5. {
  6.     return pool->flags & THRDP_STATUS_STOPPING;
  7. }

  9. static void* thrdp_process_func(void* arg)
  10. {
  11.     struct thread_pool* pool = (struct thread_pool*)arg;
  12.     struct thread_job* job;

  14.     while (1)
  15.     {
  16.         if (thrdp_stopping(pool))
  17.         {
  18.             pthread_exit(NULL);
  19.         }

  21.         pthread_mutex_lock(&(pool->qlock));
  22.         while (list_empty(&pool->job_queue))
  23.         {
  24.             // nothing to process, wait until job come in
  25.             printf("thread<%u> get none job, hangup...\n", (unsigned int)pthread_self() );
  26.             pthread_cond_wait(&(pool->job_dequeue_valid), &(pool->qlock));

  28.             if (thrdp_stopping(pool))
  29.             {
  30.                 pthread_exit(NULL);
  31.             }

  33.         }
  34.         job = list_first_entry(&pool->job_queue, struct thread_job, list);
  35.         list_del(&job->list);
  36.         pthread_mutex_unlock(&(pool->qlock));
  37.         job->jfunc(job);
  38.     }
  39. }


  42. struct thread_pool* THRDP_new(unsigned int thread_num, unsigned int max_job_num)
  43. {
  44.     struct thread_pool* pool = NULL;
  45.     struct thread_job* job;
  46.     struct thread_job* tmp;
  47.     int i;

  49.     if(thread_num < 2)
  50.     {
  51.         thread_num = 2;
  52.     }

  54.     if (max_job_num < thread_num)
  55.     {
  56.         max_job_num = thread_num;
  57.     }

  59.     pool = malloc(sizeof(struct thread_pool));
  60.     if (NULL == pool)
  61.     {
  62.         printf("failed to malloc thread_pool!\n");
  63.         return NULL;
  64.     }
  65.     memset(pool, 0, sizeof(struct thread_pool) );
  66.     pool->thread_num = thread_num;
  67.     pool->max_job_num = max_job_num;

  69.     INIT_LIST_HEAD(&pool->job_queue);
  70.     INIT_LIST_HEAD(&pool->free_job_queue);

  72.     if (pthread_mutex_init(&(pool->qlock), NULL))
  73.     {
  74.         printf("failed to init job queue mutex lock!\n");
  75.         goto ERROR_OUT;
  76.     }
  77.     if (pthread_mutex_init(&(pool->fqlock), NULL))
  78.     {
  79.         printf("failed to init free job queue mutex lock!\n");
  80.         goto ERROR_OUT1;
  81.     }
  82.     if (pthread_cond_init(&(pool->job_dequeue_valid), NULL))
  83.     {
  84.         printf("failed to init cond job_dequeue_valid!\n");
  85.         goto ERROR_OUT2;
  86.     }
  87.     if (pthread_cond_init(&(pool->job_enqueue_valid), NULL))
  88.     {
  89.         printf("failed to init cond job_enqueue_valid!\n");
  90.         goto ERROR_OUT3;
  91.     }

  93.     for (i = 0; i < max_job_num; i++)
  94.     {
  95.         job = (struct thread_job*)malloc(sizeof(struct thread_job));
  96.         if (NULL == job)
  97.         {
  98.             goto ERROR_OUT4;
  99.         }
  100.         memset(job, 0, sizeof(struct thread_job));
  101.         list_add(&job->list, &pool->free_job_queue);
  102.     }

  104.     pool->pthreads = malloc(sizeof(pthread_t) * thread_num);
  105.     if (NULL == pool->pthreads)
  106.     {
  107.         printf("failed to malloc pthreads!\n");
  108.         goto ERROR_OUT4;
  109.     }
  110.     memset(pool->pthreads, 0, sizeof(pthread_t) * thread_num);


  113.     for (i = 0; i < pool->thread_num; ++i)
  114.     {
  115.         pthread_create(&(pool->pthreads[i]), NULL, thrdp_process_func, (void*)pool);
  116.     }

  118.     printf("create pool done\n");
  119.     return pool;

  121. ERROR_OUT4:
  122.     if (!list_empty(&pool->free_job_queue))
  123.     {
  124.         list_for_each_entry_safe(job, tmp, &pool->free_job_queue, list)
  125.         {
  126.             list_del(&job->list);
  127.             free(job);
  128.         }
  129.     }
  130.     pthread_cond_destroy(&(pool->job_enqueue_valid));
  131. ERROR_OUT3:
  132.     pthread_cond_destroy(&(pool->job_dequeue_valid));
  133. ERROR_OUT2:
  134.     pthread_mutex_destroy(&(pool->fqlock));
  135. ERROR_OUT1:
  136.     pthread_mutex_destroy(&(pool->qlock));
  137. ERROR_OUT:
  138.     free(pool);
  139.     return NULL;

  141. }

  143. void THRDP_destroy(struct thread_pool* pool)
  144. {
  145.     struct thread_job* job;
  146.     struct thread_job* tmp;
  147.     int i;

  149.     if (!pool)
  150.     {
  151.         return;
  152.     }
  153.     pthread_mutex_lock(&(pool->qlock));
  154.     if (thrdp_stopping(pool)) //线程池已经退出了，就直接返回
  155.     {
  156.         pthread_mutex_unlock(&(pool->qlock));
  157.         return;
  158.     }

  160.     // 設置stop標記，喚醒阻塞的線程，使其退出
  161.     pool->flags |= THRDP_STATUS_STOPPING;
  162.     pthread_mutex_unlock(&(pool->qlock));

  164.     pthread_cond_broadcast(&(pool->job_dequeue_valid)); //唤醒线程池中正在阻塞的线程
  165.     pthread_cond_broadcast(&(pool->job_enqueue_valid)); //唤醒添加任务的threadpool_add_job函数

  167.     for (i = 0; i < pool->thread_num; ++i)
  168.     {
  169.         pthread_join(pool->pthreads[i], NULL); //等待线程池的所有线程执行完毕
  170.     }
  171.     free(pool->pthreads);

  173.     // all threads had done, no one using jobs, just free them
  174.     if (!list_empty(&pool->job_queue))
  175.     {
  176.         list_for_each_entry_safe(job, tmp, &pool->job_queue, list)
  177.         {
  178.             list_del(&job->list);
  179.             free(job);
  180.         }
  181.     }

  183.     if (!list_empty(&pool->free_job_queue))
  184.     {
  185.         list_for_each_entry_safe(job, tmp, &pool->free_job_queue, list)
  186.         {
  187.             list_del(&job->list);
  188.             free(job);
  189.         }
  190.     }

  192.     pthread_mutex_destroy(&(pool->qlock));
  193.     pthread_mutex_destroy(&(pool->fqlock));
  194.     pthread_cond_destroy(&(pool->job_dequeue_valid));
  195.     pthread_cond_destroy(&(pool->job_enqueue_valid));
  196. }

  198. /**
  199. * get a free JOB
  200. * if all jobs busy, wait util someone be free
  201. */
  202. struct thread_job* THRDP_job_get(struct thread_pool* pool)
  203. {
  204.     struct thread_job* job;

  206.     if (!pool || thrdp_stopping(pool))
  207.     {
  208.         return NULL;
  209.     }

  211.     pthread_mutex_lock(&(pool->fqlock));
  212.     while (list_empty(&pool->free_job_queue))
  213.     {
  214.         pthread_cond_wait(&(pool->job_enqueue_valid), &(pool->fqlock));
  215.         if (thrdp_stopping(pool))
  216.         {
  217.             pthread_mutex_unlock(&(pool->fqlock));
  218.             return NULL;
  219.         }

  221.     }
  222.     job = list_first_entry(&pool->free_job_queue, struct thread_job, list);
  223.     list_del(&job->list);
  224.     pthread_mutex_unlock(&(pool->fqlock));
  225.     job->pool = pool;
  226.     return job;
  227. }


  230. /**
  231. * put back a used job back to free_job_queue
  232. *
  233. * this will free the job data which MUST allocated dynamically
  234. */
  235. void THRDP_job_put(struct thread_pool* pool, struct thread_job* job)
  236. {
  237.     int notify = 0;
  238.     if (!pool || !job )
  239.     {
  240.         return;
  241.     }
  242.     if (job->data)
  243.     {
  244.         free(job->data);
  245.         job->data = NULL;
  246.     }

  248.     memset(job, 0, sizeof(struct thread_job));

  250.     pthread_mutex_lock(&(pool->fqlock));
  251.     notify = list_empty(&pool->free_job_queue);
  252.     list_add_tail(&job->list, &pool->free_job_queue);
  253.     pthread_mutex_unlock(&(pool->fqlock));

  255.     if (notify)
  256.     {
  257.         printf("put back %p to free queue, notify wait jos tasks\n", job);
  258.         pthread_cond_broadcast(&(pool->job_enqueue_valid));
  259.     }
  260. }

  262. /**
  263. * add a job to thread-pool for schedule
  264. */
  265. void THRDP_job_add(struct thread_pool* pool, struct thread_job* job)
  266. {
  267.     int notify = 0;

  269.     if (!pool || !job )
  270.     {
  271.         return;
  272.     }

  274.     pthread_mutex_lock(&(pool->qlock));
  275.     notify = list_empty(&pool->job_queue);
  276.     list_add_tail(&job->list, &pool->job_queue);
  277.     pthread_mutex_unlock(&(pool->qlock));

  279.     if (notify)
  280.     {
  281.         // notify somebody to handle the new job
  282.         printf("add new job, notify someone to handle it\n");
  283.         pthread_cond_broadcast(&(pool->job_dequeue_valid));
  284.     }
  285. }

测试文件thrdp_test.c


点击(此处)折叠或打开

  1. #include <stdio.h>

  3. #include <string.h>
  4. #include <stdlib.h>
  5. #include <syslog.h>
  6. #include <sys/socket.h>
  7. #include <sys/types.h>
  8. #include <signal.h>
  9. #include <getopt.h>
  10. #include <netinet/in.h>
  11. #include <arpa/inet.h>
  12. #include <string.h>
  13. #include <sys/select.h>
  14. #include <sys/un.h>
  15. #include <stddef.h>
  16. #include <unistd.h>
  17. #include <errno.h>
  18. #include <sys/stat.h>
  19. #include <fcntl.h>
  20. #include <sys/wait.h>
  21. #include <unistd.h>
  22. #include <uuid/uuid.h>
  23. #include <sys/time.h>

  25. #include "threadpool.h"

  27. #define SRV_PORT 9966
  28. #define UDP_PKG_MAX_LEN 4096

  30. struct pkg_desc
  31. {
  32.     int srvfd;
  33.     struct sockaddr_in cliaddr;
  34.     char data[UDP_PKG_MAX_LEN];
  35. };


  38. void pkg_process(void* job)
  39. {
  40.     struct thread_job* pjob = (struct thread_job*) job;
  41.     struct pkg_desc* ppkg = (struct pkg_desc*) pjob->data;

  43.     printf("************* process in %u ******************\n", (unsigned int)pthread_self());
  44.     printf("content:: %s\n", ppkg->data);
  45.     sendto(ppkg->srvfd, ppkg->data, strlen(ppkg->data), 0, (struct sockaddr*)&ppkg->cliaddr, sizeof(ppkg->cliaddr));
  46.     THRDP_job_put(pjob->pool, pjob);
  47. }


  50. void start_server(void)
  51. {
  52.     struct sockaddr_in srvin, cliin;
  53.     socklen_t addrlen;
  54.     int cmd_mgmt_fd;
  55.     struct thread_pool* pool = NULL;


  58.     cmd_mgmt_fd = socket(AF_INET, SOCK_DGRAM, 0);
  59.     if(cmd_mgmt_fd < 0)
  60.     {
  61.         printf("start_cmd_mgmt_process socket fd failed.\n");
  62.         return;
  63.     }

  65.     memset(&srvin, 0, sizeof(struct sockaddr_in));
  66.     srvin.sin_family = AF_INET;
  67.     srvin.sin_addr.s_addr = inet_addr("127.0.0.1");
  68.     srvin.sin_port = htons(SRV_PORT);

  70.     if(bind(cmd_mgmt_fd, (struct sockaddr*)&srvin, sizeof(struct sockaddr_in)) < 0)
  71.     {
  72.         printf("bind error");
  73.     }


  76.     if ((pool = THRDP_new(2, 8)) == NULL)
  77.     {
  78.         printf("create thread pool error\n");
  79.         return;
  80.     }

  82.     while(1)
  83.     {
  84.         struct pkg_desc* ppkg = NULL;
  85.         struct thread_job* job = THRDP_job_get(pool);
  86.         int count;

  88.         if (NULL == job)
  89.         {
  90.             printf("get thread process task error\n");
  91.             return;
  92.         }
  93.         ppkg = (struct pkg_desc*) malloc(sizeof(struct pkg_desc));

  95.         if (NULL == ppkg)
  96.         {
  97.             printf("malloc pkg error\n");
  98.             return;
  99.         }
  100.         memset(ppkg, 0, sizeof(struct pkg_desc));


  103.         addrlen = sizeof(struct sockaddr_in);
  104.         count = recvfrom(cmd_mgmt_fd, ppkg->data, UDP_PKG_MAX_LEN, 0, (struct sockaddr*)&ppkg->cliaddr, &addrlen);
  105.         if(count <= 0)
  106.         {
  107.             printf("recvform error\n");
  108.             continue;
  109.         }

  111.         ppkg->srvfd = cmd_mgmt_fd;
  112.         job->data = ppkg;
  113.         job->jfunc = pkg_process;
  114.         THRDP_job_add(pool, job);
  115.     }
  116. }


  119. void start_client(int client_id)
  120. {
  121.     struct sockaddr_in srvaddr;
  122.     struct sockaddr_in peeraddr;
  123.     int len = sizeof(peeraddr);
  124.     int fd;
  125.     char ctnbuf[512] = {0};
  126.     char rcvbuf[512] = {0};
  127.     unsigned int i = 0;
  128.     fd = socket(AF_INET, SOCK_DGRAM, 0);
  129.     if(fd < 0)
  130.     {
  131.         printf("start_client socket fd failed.\n");
  132.         return;
  133.     }

  135.     memset(&srvaddr, 0, sizeof(struct sockaddr_in));
  136.     srvaddr.sin_family = AF_INET;
  137.     srvaddr.sin_addr.s_addr = inet_addr("127.0.0.1");
  138.     srvaddr.sin_port = htons(SRV_PORT);


  141.     while(1)
  142.     {

  144.         memset(ctnbuf, 0, 512);
  145.         sprintf(ctnbuf, "%d say just test:%u", client_id, i++);

  147.         sendto(fd, ctnbuf, strlen(ctnbuf), 0, (struct sockaddr*)&srvaddr, sizeof(srvaddr));
  148.         memset(rcvbuf, 0, 512);
  149.         len = sizeof(peeraddr);
  150.         recvfrom(fd, rcvbuf, 512, 0, (struct sockaddr*)&peeraddr, &len);
  151.         printf("recv content:%s\n", rcvbuf);
  152.     }
  153. }

  155. int main(int argc, char** argv)
  156. {
  157.     if (argc < 2)
  158.     {
  159.         printf("usage : %s [-s/-c] [cid]", argv[0]);
  160.         return 0;
  161.     }
  162.     if (!strcmp(argv[1], "-s"))
  163.     {
  164.         start_server();
  165.     }
  166.     else
  167.     {
  168.         start_client(atoi(argv[2]));
  169.     }
  170.     return 0;

  172. }

测试是一个简单的udp回射服务器模型，用线程池的方式做服务端并发。

编译： gcc -g -o thrdp *.c

启动server端： ./thrdp -s

多终端启动client端： ./thrdp -c 100，./thrdp -c 102， ...

运行截图如下




阅读(2461) | 评论(0) | 转发(0) |
0

上一篇：背景知识——EJB

下一篇：DHCP 服务探测

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6