线程池原理及C语言实现线程池

备注：该线程池源码参考自传直播客培训视频配套资料；
源码：https://pan.baidu.com/s/1zWuoE3q0KT5TUjmPKTb1lw 密码：pp42
引言：线程池是一种多线程处理形式，大多用于高并发服务器上，它能合理有效的利用高并发服务器上的线程资源；

在Unix网络编程中，线程与进程用于处理各项分支子功能，我们通常的操作是：接收消息 ==> 消息分类 ==> 线程创建 ==> 传递消息到子线程 ==> 线程分离 ==> 在子线程中执行任务 ==> 任务结束退出；
对大多数小型局域网的通信来说，上述方法足够满足需求；但当我们的通信范围扩大到广域网或大型局域网通信中时，我们将面临大量消息频繁请求服务器；在这种情况下，创建与销毁线程都已经成为一种奢侈的开销，特别对于嵌入式服务器来说更应保证内存资源的合理利用；
因此，线程池技术应运而生；线程池允许一个线程可以多次复用，且每次复用的线程内部的消息处理可以不相同，将创建与销毁的开销省去而不必来一个请求开一个线程；

结构讲解：
线程池是一个抽象的概念，其内部由任务队列，一堆线程，管理者线程组成；
这里写图片描述


我们将以上图为例，实现一个最基础的线程池，接下来将分部分依次讲解；讲解顺序为：1.线程池总体结构 2.线程数组 3.任务队列 4.管理者线程 5.使用线程池接口的例子
一、线程池总体结构

这里讲解线程池在逻辑上的结构体；看下方代码，该结构体threadpool_t中包含线程池状态信息，任务队列信息以及多线程操作中的互斥锁；在任务结构体中包含了一个可以放置多种不同任务函数的函数指针，一个传入该任务函数的void*类型的参数；
注意：在使用时需要将你的消息分类处理函数装入任务的(*function);然后放置到任务队列并通知空闲线程；

线程池状态信息：描述当前线程池的基本信息，如是否开启、最小线程数、最大线程数、存活线程数、忙线程数、待销毁线程数等… …
任务队列信息：描述当前任务队列基本信息，如最大任务数、队列不为满条件变量、队列不为空条件变量等… …
多线程互斥锁：保证在同一时间点上只有一个线程在任务队列中取任务并修改任务队列信息、修改线程池信息；
函数指针：在打包消息阶段，将分类后的消息处理函数放在(*function);
void*类型参数：用于传递消息处理函数需要的信息；

/*任务*/
typedef struct {
   void *(*function)(void *);
   void *arg;
} threadpool_task_t;

/*线程池管理*/
struct threadpool_t{
   pthread_mutex_t lock;                 /* 锁住整个结构体 */
   pthread_mutex_t thread_counter;       /* 用于使用忙线程数时的锁 */
   pthread_cond_t  queue_not_full;       /* 条件变量，任务队列不为满 */
   pthread_cond_t  queue_not_empty;      /* 任务队列不为空 */

   pthread_t *threads;                   /* 存放线程的tid,实际上就是管理了线 数组 */
   pthread_t admin_tid;                  /* 管理者线程tid */
   threadpool_task_t *task_queue;        /* 任务队列 */

   /*线程池信息*/
   int min_thr_num;                      /* 线程池中最小线程数 */
   int max_thr_num;                      /* 线程池中最大线程数 */
   int live_thr_num;                     /* 线程池中存活的线程数 */
   int busy_thr_num;                     /* 忙线程，正在工作的线程 */
   int wait_exit_thr_num;                /* 需要销毁的线程数 */

   /*任务队列信息*/
   int queue_front;                      /* 队头 */
   int queue_rear;                       /* 队尾 */
   int queue_size;

   /* 存在的任务数 */
   int queue_max_size;                   /* 队列能容纳的最大任务数 */
   /*线程池状态*/
   int shutdown;                         /* true为关闭 */
};

   

**/*创建线程池*/**
threadpool_t *
threadpool_create(int min_thr_num, int max_thr_num, int queue_max_size)
{               /*  最小线程数           最大线程数         最大任务数*/
   int i;
   threadpool_t *pool = NULL;
   do
   {
      /* 线程池空间开辟 */
      if ((pool=(threadpool_t *)malloc(sizeof(threadpool_t))) == NULL)
      {
        printf("malloc threadpool false; \n");
        break;   
      }
      /*信息初始化*/
      pool->min_thr_num = min_thr_num;
      pool->max_thr_num = max_thr_num;
      pool->busy_thr_num = 0;
      pool->live_thr_num = min_thr_num;
      pool->wait_exit_thr_num = 0;
      pool->queue_front = 0;
      pool->queue_rear = 0;
      pool->queue_size = 0;
      pool->queue_max_size = queue_max_size;
      pool->shutdown = false;

      /* 根据最大线程数，给工作线程数组开空间，清0 */
      pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num);
      if (pool->threads == NULL)
      {
         printf("malloc threads false;\n");
         break;
      }
      memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num);

      /* 队列开空间 */
      pool->task_queue =
      (threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size);
      if (pool->task_queue == NULL)
      {
         printf("malloc task queue false;\n");
         break;
      }

      /* 初始化互斥锁和条件变量 */
      if ( pthread_mutex_init(&(pool->lock), NULL) != 0           ||
           pthread_mutex_init(&(pool->thread_counter), NULL) !=0  ||
       pthread_cond_init(&(pool->queue_not_empty), NULL) !=0  ||
       pthread_cond_init(&(pool->queue_not_full), NULL) !=0)
      {
         printf("init lock or cond false;\n");
         break;
      }

      /* 启动min_thr_num个工作线程 */
      for (i=0; i<min_thr_num; i++)
      {
         /* pool指向当前线程池  threadpool_thread函数在后面讲解 */
         pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
         printf("start thread 0x%x... \n", (unsigned int)pool->threads[i]);
      }
      /* 管理者线程 admin_thread函数在后面讲解 */
      pthread_create(&(pool->admin_tid), NULL, admin_thread, (void *)pool);

      return pool;
   } while(0);

   /* 释放pool的空间 */
   threadpool_free(pool);
   return NULL;
}

   

二、线程数组

线程数组实际上是在线程池初始化时开辟的一段存放一堆线程tid的空间，在逻辑上形成一个池，里面放置着提前创建的线程；这段空间中包含了正在工作的线程，等待工作的线程(空闲线程)，等待被销毁的线程，申明但没有初始化的线程空间；
这里写图片描述

/*工作线程*/
void *
threadpool_thread(void *threadpool)
{
  threadpool_t *pool = (threadpool_t *)threadpool;
  threadpool_task_t task;

  while (true)
  {
    pthread_mutex_lock(&(pool->lock));

    /* 无任务则阻塞在 “任务队列不为空” 上，有任务则跳出 */
    while ((pool->queue_size == 0) && (!pool->shutdown))
    {
       printf("thread 0x%x is waiting \n", (unsigned int)pthread_self());
       pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));

       /* 判断是否需要清除线程,自杀功能 */
       if (pool->wait_exit_thr_num > 0)
       {
          pool->wait_exit_thr_num--;
          /* 判断线程池中的线程数是否大于最小线程数，是则结束当前线程 */
          if (pool->live_thr_num > pool->min_thr_num)
          {
             printf("thread 0x%x is exiting \n", (unsigned int)pthread_self());
             pool->live_thr_num--;
             pthread_mutex_unlock(&(pool->lock));
             pthread_exit(NULL);//结束线程
          }
       }
    }

    /* 线程池开关状态 */
    if (pool->shutdown) //关闭线程池
    {
       pthread_mutex_unlock(&(pool->lock));
       printf("thread 0x%x is exiting \n", (unsigned int)pthread_self());
       pthread_exit(NULL); //线程自己结束自己
    }

    //否则该线程可以拿出任务
    task.function = pool->task_queue[pool->queue_front].function; //出队操作
    task.arg = pool->task_queue[pool->queue_front].arg;

    pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size;  //环型结构
    pool->queue_size--;

    //通知可以添加新任务
    pthread_cond_broadcast(&(pool->queue_not_full));

    //释放线程锁
    pthread_mutex_unlock(&(pool->lock));

    //执行刚才取出的任务
    printf("thread 0x%x start working \n", (unsigned int)pthread_self());
    pthread_mutex_lock(&(pool->thread_counter));            //锁住忙线程变量
    pool->busy_thr_num++;
    pthread_mutex_unlock(&(pool->thread_counter));

    (*(task.function))(task.arg);                           //执行任务

    //任务结束处理
    printf("thread 0x%x end working \n", (unsigned int)pthread_self());
    pthread_mutex_lock(&(pool->thread_counter));
    pool->busy_thr_num--;
    pthread_mutex_unlock(&(pool->thread_counter));
  }

  pthread_exit(NULL);
}

   

三、任务队列

任务队列的存在形式与线程数组相似；在线程池初始化时根据传入的最大任务数开辟空间；当服务器前方后请求到来后，分类并打包消息成为任务，将任务放入任务队列并通知空闲线程来取；不同之处在于任务队列有明显的先后顺序，先进先出；而线程数组中的线程则是一个竞争关系去拿到互斥锁争取任务；
这里写图片描述

/*向线程池的任务队列中添加一个任务*/
int
threadpool_add_task(threadpool_t *pool, void *(*function)(void *arg), void *arg)
{
   pthread_mutex_lock(&(pool->lock));

   /*如果队列满了,调用wait阻塞*/
   while ((pool->queue_size == pool->queue_max_size) && (!pool->shutdown))
      pthread_cond_wait(&(pool->queue_not_full), &(pool->lock));

   /*如果线程池处于关闭状态*/
   if (pool->shutdown)
   {
      pthread_mutex_unlock(&(pool->lock));
      return -1;
   }

   /*清空工作线程的回调函数的参数arg*/
   if (pool->task_queue[pool->queue_rear].arg != NULL)
   {
      free(pool->task_queue[pool->queue_rear].arg);
      pool->task_queue[pool->queue_rear].arg = NULL;
   }

   /*添加任务到任务队列*/
   pool->task_queue[pool->queue_rear].function = function;
   pool->task_queue[pool->queue_rear].arg = arg;
   pool->queue_rear = (pool->queue_rear + 1) % pool->queue_max_size;  /* 逻辑环  */
   pool->queue_size++;

   /*添加完任务后,队列就不为空了,唤醒线程池中的一个线程*/
   pthread_cond_signal(&(pool->queue_not_empty));
   pthread_mutex_unlock(&(pool->lock));

   return 0;
}

   

四、管理者线程

作为线程池的管理者，该线程的主要功能包括：检查线程池内线程的存活状态，工作状态；负责根据服务器当前的请求状态去动态的增加或删除线程，保证线程池中的线程数量维持在一个合理高效的平衡上；
说到底，它就是一个单独的线程，定时的去检查，根据我们的一个维持平衡算法去增删线程；

/*管理线程*/
void *
admin_thread(void *threadpool)
{
   int i;
   threadpool_t *pool = (threadpool_t *)threadpool;
   while (!pool->shutdown)
   {
      printf("admin -----------------\n");
      sleep(DEFAULT_TIME);                             /*隔一段时间再管理*/
      pthread_mutex_lock(&(pool->lock));               /*加锁*/
      int queue_size = pool->queue_size;               /*任务数*/
      int live_thr_num = pool->live_thr_num;           /*存活的线程数*/
      pthread_mutex_unlock(&(pool->lock));             /*解锁*/

      pthread_mutex_lock(&(pool->thread_counter));
      int busy_thr_num = pool->busy_thr_num;           /*忙线程数*/  
      pthread_mutex_unlock(&(pool->thread_counter));

      printf("admin busy live -%d--%d-\n", busy_thr_num, live_thr_num);
      /*创建新线程 实际任务数量大于 最小正在等待的任务数量，存活线程数小于最大线程数*/
      if (queue_size >= MIN_WAIT_TASK_NUM && live_thr_num <= pool->max_thr_num)
      {
         printf("admin add-----------\n");
         pthread_mutex_lock(&(pool->lock));
         int add=0;

         /*一次增加 DEFAULT_THREAD_NUM 个线程*/
         for (i=0; i<pool->max_thr_num && add<DEFAULT_THREAD_NUM
              && pool->live_thr_num < pool->max_thr_num; i++)
         {
            if (pool->threads[i] == 0 || !is_thread_alive(pool->threads[i]))
           {
              pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
              add++;
              pool->live_thr_num++;
              printf("new thread -----------------------\n");
           }
         }

         pthread_mutex_unlock(&(pool->lock));
      }

      /*销毁多余的线程 忙线程x2 都小于 存活线程，并且存活的大于最小线程数*/
      if ((busy_thr_num*2) < live_thr_num  &&  live_thr_num > pool->min_thr_num)
      {
         // printf("admin busy --%d--%d----\n", busy_thr_num, live_thr_num);
         /*一次销毁DEFAULT_THREAD_NUM个线程*/
         pthread_mutex_lock(&(pool->lock));
         pool->wait_exit_thr_num = DEFAULT_THREAD_NUM;
         pthread_mutex_unlock(&(pool->lock));

         for (i=0; i<DEFAULT_THREAD_NUM; i++)
        {
           //通知正在处于空闲的线程，自杀
           pthread_cond_signal(&(pool->queue_not_empty));
           printf("admin cler --\n");
        }
      }

   }

   return NULL;


/*线程是否存活*/
int
is_thread_alive(pthread_t tid)
{
   int kill_rc = pthread_kill(tid, 0);     //发送0号信号，测试是否存活
   if (kill_rc == ESRCH)  //线程不存在
   {
      return false;
   }
   return true;
}

   

五、释放

/*释放线程池*/
int
threadpool_free(threadpool_t *pool)
{
   if (pool == NULL)
     return -1;
   if (pool->task_queue)
      free(pool->task_queue);
   if (pool->threads)
   {
      free(pool->threads);
      pthread_mutex_lock(&(pool->lock));               /*先锁住再销毁*/
      pthread_mutex_destroy(&(pool->lock));
      pthread_mutex_lock(&(pool->thread_counter));
      pthread_mutex_destroy(&(pool->thread_counter));
      pthread_cond_destroy(&(pool->queue_not_empty));
      pthread_cond_destroy(&(pool->queue_not_full));
   }
   free(pool);
   pool = NULL;

   return 0;
}

  

/*销毁线程池*/
int
threadpool_destroy(threadpool_t *pool)
{
   int i;
   if (pool == NULL)
   {
     return -1;
   }
   pool->shutdown = true;

   /*销毁管理者线程*/
   pthread_join(pool->admin_tid, NULL);

   //通知所有线程去自杀(在自己领任务的过程中)
   for (i=0; i<pool->live_thr_num; i++)
   {
     pthread_cond_broadcast(&(pool->queue_not_empty));
   }

   /*等待线程结束 先是pthread_exit 然后等待其结束*/
   for (i=0; i<pool->live_thr_num; i++)
   {
     pthread_join(pool->threads[i], NULL);
   }

   threadpool_free(pool);
   return 0;
}

  

六、接口

   /* 线程池初始化，其管理者线程及工作线程都会启动 */
    threadpool_t *thp = threadpool_create(10, 100, 100);
    printf("threadpool init ... ... \n");

   /* 接收到任务后添加 */
   threadpool_add_task(thp, do_work, (void *)p);

   // ... ...

   /* 销毁 */
   threadpool_destroy(thp)；

  

 

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/384931.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

iptables 的mangle表

iptables 的mangle表

mangle表的主要功能是根据规则修改数据包的一些标志位&#xff0c;以便其他规则或程序可以利用这种标志对数据包进行过滤或策略路由。 内网的客户机通过Linux主机连入Internet&#xff0c;而Linux主机与Internet连接时有两条线路&#xff0c;它们的网关如图所示。现要求对内网进…
阅读更多...
Linux之静态库

Linux之静态库

命名规则&#xff1a; lib 库的名字 .a 制作步骤 生成对应.o文件 .c .o 将生成的.o文件打包 ar rcs 静态库的名字&#xff08;libMytest.a&#xff09; 生成的所有的.o 发布和使用静态库&#xff1a; 1&#xff09; 发布静态 2&#xff09; 头文件 文件如下图所示&…
阅读更多...
Linux之动态库

Linux之动态库

命令规则 lib 名字 .so 制作步骤 1&#xff09;生成与位置无关的代码&#xff08;生成与位置无关的代码&#xff09; 2&#xff09;将.o打包成共享库&#xff08;动态库&#xff09; 发布和使用共享库 动态库运行原理&#xff1a; 生成动态库&#xff1a; gcc -fPIC -c *.c -…
阅读更多...
linux下源码安装vsftpd-3.0.2

linux下源码安装vsftpd-3.0.2

1&#xff09;在http://vsftpd.beasts.org/网站中查找并下载 vsftpd-3.0.2.tar.gz源码包 2)如果自己的机器上安装有yum可以用yum grouplist | less指令查看以下开发环境&#xff0c;当然这一步不做也行 3&#xff09;拆解源码包 4&#xff09;查看源码包 5&#xff09;编辑…
阅读更多...
Gdb 调试core文件详解

Gdb 调试core文件详解

一&#xff0c;什么是coredump 我们经常听到大家说到程序core掉了&#xff0c;需要定位解决&#xff0c;这里说的大部分是指对应程序由于各种异常或者bug导致在运行过程中异常退出或者中止&#xff0c;并且在满足一定条件下&#xff08;这里为什么说需要满足一定的条件呢&#…
阅读更多...
Windows下编译openssl库

Windows下编译openssl库

1、概述 OpenSSL是一个开放源代码的软件库包&#xff0c;它实现了 SSL&#xff08;Secure SocketLayer&#xff09;和 TLS&#xff08;Transport Layer Security&#xff09;协议&#xff0c;所以应用程序可以使用这个包来进行安全通信&#xff0c;避免窃听&#xff0c;同时确…
阅读更多...
windows环境下C语言socket编程

windows环境下C语言socket编程

最近由于实验需要&#xff0c;要求写一个c程序与java程序通信的软件&#xff0c;为了测试首先写了一个windows环境下c语言的socket&#xff08;tcp&#xff09;通信程序。 首先socket通信的步骤&#xff1a; 图一 socket通信步骤&#xff08;转载) 图二 三次握手协议&…
阅读更多...
Makefile（三）

Makefile（三）

在平时使用中&#xff0c;可以使用以下的makefile来编译单独的代码 src $(wildcard *.c) obj $(patsubst %.c, %.o, $(src))CC gcc CFLAGS -Wall -gall:$(target)$(target):%:%.c$(CC) $< -o $ $(CFLAGS).PHONY: clean all clean:-rm -rf $(target) 使用方法就是make 后…
阅读更多...
数据类型（C++）

数据类型（C++）

不同系统会有不同差异&#xff1a; 类型 位(byte) 范围 char 1 -128—127 or 0 – 255 unsigned char 1 0 – 255 signed int 1 -128—127 int 4 -2^31 – 2^32-1 unsigned int 4 0 – 2^32 signed int 4 -2^31 – 2^32-1 short int 2 2^15 – 2^15-1 …
阅读更多...
dup2函数

dup2函数

将当前系统中的进程信息打印到文件中 命令行&#xff1a;ps aux > out 将ps得到的信息重定向到out文件中 使用dup2文件在程序中完成。 int dup2(int oldfd,int newfd); /*** dup2.c ***/ #include<stdio.h> #include<fcntl.h> #include<unistd.h> #includ…
阅读更多...
内核实现信号捕捉原理

内核实现信号捕捉原理

信号捕捉特性 进程正常运行时&#xff0c;默认PCB中有一个信号屏蔽字&#xff0c;假定为☆&#xff0c;它决定了进程自动屏蔽哪些信号。当注册了某个信号捕捉函数&#xff0c;捕捉到该信号以后&#xff0c;要调用该函数。而该函数有可能执行很长时间&#xff0c;在这期间所屏蔽…
阅读更多...
C++预处理器

C++预处理器

预处理器是一些指令&#xff0c;指示表一起在实际编译之前所需要完成的预处理。 所有的预处理器指令都是以井号(#)开头&#xff0c;只有空格字符可以出现在预指令处理之前。预处理指令不是C语句&#xff0c;所以他们不会以分号(;)结尾。 #define预处理 #define预处理指令用于创…
阅读更多...
Python变量类型

Python变量类型

变量存储在内存中的值&#xff0c;这就意味着在创建变量时会在内存开辟一个空间。 基于变量的数据类型&#xff0c;解析器会分配指定内存&#xff0c;并决定什么数据可以被存储在内存中。 因此变量可以指定不同的数据类型&#xff0c;这些变量可以存储整数、小数、或字符。 变量…
阅读更多...
Python3字符串

Python3字符串

字符串是Python中最常用的数据类型&#xff0c;可以使用单引号或双引号来创建字符串 创建字符串很简单&#xff0c;为变量分配一个值即可。 val1 ‘hello world’ var2 “Runoob” Python访问字符串的值 Python不支持单字符类型&#xff0c;单字符在Python中也是作为 一个字符…
阅读更多...
服务器客户端编程

服务器客户端编程

server 下面通过最简单的客户端/服务器程序的实例来学习socket API。 server.c的作用是从客户端读字符&#xff0c;然后将每个字符转换为大写并回送给客户端。 #include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #incl…
阅读更多...
Python3元组

Python3元组

Python的元组与列表相似&#xff0c;不同之处在于元组的元素不能修改 元组使用小括号&#xff0c;列表使用方括号 元组创建很简单&#xff0c;只需要在括号中添加元素&#xff0c;并使用逗号隔开即可。 创建空元组 tup1 (); tup2 (1,) 元组只包含一个元素时&#xff0c;需要在…
阅读更多...
Python3字典

Python3字典

字典是另一种可变容器模型&#xff0c;可存储任意类型的对象。 字典的每个键值(key>value)对用冒号分隔&#xff0c;每个对之间用逗号分隔&#xff0c;整个字典包括在花括号里&#xff0c;格式如下 d {key1 : value,key2 : value2} 键必须是唯一&#xff0c;但值则不必。 值…
阅读更多...
多进程服务器

多进程服务器

注意&#xff1a;包含了“wrap.c” 和“wrap.h”文件在上篇博客中 /*** server.c ***/ #include<stdio.h> #include<string.h> #include<netinet/in.h> #include<arpa/inet.h> #include<signal.h> #include<sys/wait.h> #include<ctype…
阅读更多...
Python3集合

Python3集合

集合&#xff08;set&#xff09;是一个无序的不重复元素序列。 可以使用大括号{ } 或set&#xff08;&#xff09;函数来创建集合&#xff0c;注意&#xff1a;创建一个空集合必须用set(),{ }是用来创建一个空字典的。 创建格式&#xff1a; param {value01,value02,…} set(…
阅读更多...
Python3循环

Python3循环

Python中while语句的一般形式&#xff1a; while 判断条件: 语句 同样需要注意冒号和缩进&#xff0c;另外在Python中没有do…while循环 下面的实例计算1到100总和 ##calc.py n 100sum 0 counter 1 while counter < n:sum sum countercounter 1print("total from…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023