Linux网络通信（线程池和线程池版本的服务器代码）

乎语百科 2022-11-11 23:33:02 353 0

线程池

介绍

线程池： 一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络sockets等的数量

线程池的价值：

需要大量的线程来完成任务，且完成任务的时间比较短。可同时处理多任务，多请求。
有任务可以立即从线程池中调取线程取处理，节省了线程创建的时间
有效防止服务端线程过多而导致系统过载的问题

实现

线程池中首先需要有很多个线程，用户可以自己选择创建多少个线程。为了实现线程间的同步与互斥，还需要增加两个变量——互斥量和条件变量。我们还需要一个任务队列，主线程不断往里面塞任务，线程池的线程不断去处理。需要注意的是：这里的任务队列可以为空，但不能满，所以任务队列的容量不限定（实际场景中，任务队列容量不够就需要考虑换一台更大的服务器）

线程池的四个成员变量：

一个队列： 存放任务
线程池中线程数： 记录线程池中创建的线程数
互斥量： 一个互斥锁
条件变量： 两个条件变量

Linux网络通信（线程池和线程池版本的服务器代码）

线程池：首先需要创建几个线程，还有一个任务队列，当任务队列有任务的时候就唤醒一个正在等待的线程，让线程去执行任务，线程池中的线程执行完任务不会销毁，大大减少的cpu的消耗。

需要两个条件变量和一个互斥锁，这个互斥锁用来锁住任务队列，因为任务队列是公共资源，其次还需要两个条件变量，一个条件变量用来阻塞取任务的线程，当队列中有任务的时候，直接取任务，然后解锁，当任务队列中没有任务的时候，解锁等待条件，条件满足抢锁，取任务，解锁。另一个条件变量用来阻塞添加者进程，当任务队列满了，会让添加者进程等待，当有线程取走一个任务的时候，会唤醒添加者进程。

版本一

任务函数

这里的任务函数采用的时回调函数的方式，提高了代码的通用性，可以根据自己的需求改写任务函数

//任务回调函数
void taskRun(void *arg)
{
      PoolTask *task = (PoolTask*)arg;
      int num = task->tasknum;
      printf("task %d is runing %lu\n",num,pthread_self());

      sleep(1);
      printf("task %d is done %lu\n",num,pthread_self());
}

线程池的主要代码框架

class ThreadPool
{
public:
    //构造函数，初始化线程池
    ThreadPool(int thrnum, int maxtasknum)
    {
    }
   static void* thrRun(void* arg)
    {
    }
   //析构函数，摧毁线程池
   ~ThreadPool()
   {
   }
public:
   //添加任务到线程池
   void addtask(){}；
private:
   //任务队列相关的参数
   int max_job_num;//最大任务个数
   int job_num;//实际任务个数
   PoolTask *tasks;//任务队列数组
   int job_push;//入队位置
   int job_pop;// 出队位置

   //线程相关参数
   int thr_num;//线程池内线程个数
   pthread_t *threads;//线程池内线程数组
   int shutdown;//是否关闭线程池
   pthread_mutex_t pool_lock;//线程池的锁
   pthread_cond_t empty_task;//任务队列为空的条件
   pthread_cond_t not_empty_task;//任务队列不为空的条件
};

放任务： 主线程无脑往任务队列中塞任务，塞任务之前进行加锁，塞完任务解锁，如果任务队列已经满了，等待线程取任务，然后唤醒在条件变量下等待的队列；放入了任务就给线程发送信号，唤醒线程来取 取任务： 线程池中的线程从任务队列中取任务，需要对任务队列上锁，因为对公共资源的操作都需要上锁，如果没有任务就阻塞，等待放任务唤醒；如果取完了一个任务，就唤醒添加任务

这就是两个条件变量和一个互斥锁的用法

//添加任务到线程池
   void addtask()
   {
     pthread_mutex_lock(&(this->pool_lock));
     //实际任务总数大于最大任务个数则阻塞等待(等待任务被处理)
     while(this->max_job_num <= this->job_num)
     {
          pthread_cond_wait(&(this->empty_task),&(this->pool_lock));
     }

     int taskpos = (this->job_push++)%this->max_job_num;

     this->tasks[taskpos].tasknum = beginnum++;
     this->tasks[taskpos].arg = (void*)&this->tasks[taskpos];
     this->tasks[taskpos].task_func = taskRun;
     this->job_num++;

     pthread_mutex_unlock(&(this->pool_lock));

     pthread_cond_signal(&(this->not_empty_task));//通知包身工
   }

  //取任务
  static void* thrRun(void* arg)
    {
      ThreadPool *pool = (ThreadPool*)arg;
      int taskpos = 0;//任务位置
      PoolTask *task = new PoolTask();
      while(1)
      {
          //获取任务，先要尝试加锁
          pthread_mutex_lock(&pool->pool_lock);

          //无任务并且线程池不是要摧毁
          while(pool->job_num <= 0 && !pool->shutdown )
          {
              //如果没有任务，线程会阻塞
              pthread_cond_wait(&pool->not_empty_task,&pool->pool_lock);
          }
          if(pool->job_num)
          {
              //有任务需要处理
              taskpos = (pool->job_pop++)%pool->max_job_num;
              //printf("task out %d...tasknum===%d tid=%lu\n",taskpos,thrPool->tasks[taskpos].tasknum,pthread_self());
              //为什么要拷贝？避免任务被修改，生产者会添加任务
              memcpy(task,&pool->tasks[taskpos],sizeof(PoolTask));
              task->arg = task;
              pool->job_num--;
              //task = &thrPool->tasks[taskpos];
              pthread_cond_signal(&pool->empty_task);//通知生产者
          }

          if(pool->shutdown)
          {
              //代表要摧毁线程池，此时线程退出即可
              //pthread_detach(pthread_self());//临死前分家
              pthread_mutex_unlock(&pool->pool_lock);
              delete(task);
              pthread_exit(NULL);
          }
       //释放锁
       pthread_mutex_unlock(&pool->pool_lock);
       task->task_func(task->arg);//执行回调函数
       }
     }

整体代码：

#include<iostream>
#include<string.h>
#include<pthread.h>
#include<sys/types.h>
#include<stdio.h>
#include<unistd.h>
using namespace std;
int beginnum = 1;

class PoolTask
{
public:
    int tasknum;//模拟任务编号
    void *arg;//回调函数参数
    void (*task_func)(void *arg);//任务的回调函数
};
//任务回调函数
void taskRun(void *arg)
{
      PoolTask *task = (PoolTask*)arg;
      int num = task->tasknum;
      printf("task %d is runing %lu\n",num,pthread_self());

      sleep(1);
      printf("task %d is done %lu\n",num,pthread_self());
}
class ThreadPool
{
public:
    //构造函数，初始化线程池
    ThreadPool(int thrnum, int maxtasknum)
    {
      this->thr_num = thrnum;
      this->max_job_num = maxtasknum;
      this->shutdown = 0;//是否摧毁线程池，1代表摧毁
      this->job_push = 0;//任务队列添加的位置
      this->job_pop = 0;//任务队列出队的位置
      this->job_num = 0;//初始化的任务个数为0

      //申请最大的任务队列
      this->tasks = new PoolTask[thrnum];

      //初始化锁和条件变量
      pthread_mutex_init(&(this->pool_lock),NULL);
      pthread_cond_init(&(this->empty_task),NULL);
      pthread_cond_init(&(this->not_empty_task),NULL);

      int i = 0;
      this->threads = (pthread_t *)malloc(sizeof(pthread_t)*thrnum);//申请n个线程id的空间

      pthread_attr_t attr;
      pthread_attr_init(&attr);
      pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
      for(i = 0;i < thrnum;i++)
      {
          pthread_create(&(this->threads[i]),&attr,thrRun,this);//创建多个线程
      }
    }
   static void* thrRun(void* arg)
    {
      ThreadPool *pool = (ThreadPool*)arg;
      int taskpos = 0;//任务位置
      PoolTask *task = new PoolTask();
      while(1)
      {
          //获取任务，先要尝试加锁
          pthread_mutex_lock(&pool->pool_lock);

          //无任务并且线程池不是要摧毁
          while(pool->job_num <= 0 && !pool->shutdown )
          {
              //如果没有任务，线程会阻塞
              pthread_cond_wait(&pool->not_empty_task,&pool->pool_lock);
          }
          if(pool->job_num)
          {
              //有任务需要处理
              taskpos = (pool->job_pop++)%pool->max_job_num;
              //printf("task out %d...tasknum===%d tid=%lu\n",taskpos,thrPool->tasks[taskpos].tasknum,pthread_self());
              //为什么要拷贝？避免任务被修改，生产者会添加任务
              memcpy(task,&pool->tasks[taskpos],sizeof(PoolTask));
              task->arg = task;
              pool->job_num--;
              //task = &thrPool->tasks[taskpos];
              pthread_cond_signal(&pool->empty_task);//通知生产者
          }

          if(pool->shutdown)
          {
              //代表要摧毁线程池，此时线程退出即可
              //pthread_detach(pthread_self());//临死前分家
              pthread_mutex_unlock(&pool->pool_lock);
              delete(task);
              pthread_exit(NULL);
          }
       //释放锁
       pthread_mutex_unlock(&pool->pool_lock);
       task->task_func(task->arg);//执行回调函数
       }
     }
   //析构函数，摧毁线程池
   ~ThreadPool()
   {
     this->shutdown = 1;//关闭线程池
     pthread_cond_broadcast(&(this->not_empty_task));//诱杀

     int i = 0;
     for(i = 0; i<this->thr_num ; i++)
     {
         pthread_join(this->threads[i],NULL);
     }

     pthread_cond_destroy(&(this->not_empty_task));
     pthread_cond_destroy(&(this->empty_task));
     pthread_mutex_destroy(&(this->pool_lock));
     delete []tasks;
     tasks = NULL;
     free(this->threads);
   }

public:
   //添加任务到线程池
   void addtask()
   {
     pthread_mutex_lock(&(this->pool_lock));
     cout << "当前任务队列中任务的个数是： " <<this-> job_num <<endl;
     //实际任务总数大于最大任务个数则阻塞等待(等待任务被处理)
     while(this->max_job_num <= this->job_num)
     {
          pthread_cond_wait(&(this->empty_task),&(this->pool_lock));
     }

     int taskpos = (this->job_push++)%this->max_job_num;

     this->tasks[taskpos].tasknum = beginnum++;
     this->tasks[taskpos].arg = (void*)&this->tasks[taskpos];
     this->tasks[taskpos].task_func = taskRun;
     this->job_num++;

     pthread_mutex_unlock(&(this->pool_lock));

     pthread_cond_signal(&(this->not_empty_task));//通知包身工
   }
private:
   //任务队列相关的参数
   int max_job_num;//最大任务个数
   int job_num;//实际任务个数
   PoolTask *tasks;//任务队列数组
   int job_push;//入队位置
   int job_pop;// 出队位置

   //线程相关参数
   int thr_num;//线程池内线程个数
   pthread_t *threads;//线程池内线程数组
   int shutdown;//是否关闭线程池
   pthread_mutex_t pool_lock;//线程池的锁
   pthread_cond_t empty_task;//任务队列为空的条件
   pthread_cond_t not_empty_task;//任务队列不为空的条件
};
int main()
{
   ThreadPool *m = new ThreadPool(3,20);
   int j = 0;
   for(j=0;j<20;j++)
   {
     m->addtask();
   }
   sleep(20);
   delete m;
   m = NULL;
   system("pause");
   return EXIT_SUCCESS;
}

运行结果如下：

Linux网络通信（线程池和线程池版本的服务器代码）

可以看到线程最多处理三个任务，而任务队列中最多可以存在20个任务，当线程取走了任务之后，唤醒生产者继续添加任务。

版本二

首先封装一个任务：

class Task
{
public:
  Task(int a = 0, int b = 0)
    :_a(a)
    ,_b(b)
  {}
  void Run()
  {
    //执行的任务可以自己编写
  }
private:
  int _a;
  int _b;
};

线程池的主要代码框架（唤醒和等待操作都已经封装好）：

#define DEFAULT_MAX_PTHREAD 5

class ThreadPool
{
public:
  ThreadPool(int max_pthread = DEFAULT_MAX_PTHREAD)
    :_max_thread(max_pthread)
  {}
  ~ThreadPool()
  {
    pthread_mutex_destroy(&_mutex);
    pthread_cond_destroy(&_cond);
  }
public:
  void LockQueue()
  {
    pthread_mutex_lock(&_mutex);
  }
  void UnlockQueue()
  {
    pthread_mutex_unlock(&_mutex);
  }
  void ThreadWait()
  {
    pthread_cond_wait(&_cond, &_mutex);
  }
  void WakeUpThread()
  {
    pthread_cond_signal(&_cond);
    //pthread_cond_broadcast(&_cond);
  }
  bool IsEmpty()
  {
    return _q.empty();
  }
private:
  queue<Task*>  _q;
  int  _max_thread;
  pthread_mutex_t _mutex;
  pthread_cond_t  _cond;
};

创建多个线程 创建多个线程可以用一个循环进行创建。需要注意的是，创建一个线程还需要提供一个线程启动后要执行的函数，这个启动函数只能有一个参数。如果把这个函数设置为成员函数，那么这个函数的第一个参数默认是this指针，这样显然是不可行的，所以这里我们考虑把这个启动函数设置为静态的。但是设置为静态的成员函数又会面临一个问题：如何调用其他成员函数和成员变量？所以这里我们考虑创建线程的时候，把this指针传过去，让启动函数的arg 参数去接收即可

static void* Runtine(void* arg)
{
  pthread_detach(pthread_self());
  ThreadPool* this_p = (ThreadPool*)arg;

  while (1){
    this_p->LockQueue();
    while (this_p->IsEmpty()){
      this_p->ThreadWait();
    }
    Task* t;
    this_p->Get(t);
    this_p->UnlockQueue();
    // 解锁后处理任务
    t->Run();
    delete t;
  }
}
void ThreadPoolInit()
{
  pthread_mutex_init(&_mutex, nullptr);
  pthread_cond_init(&_cond, nullptr);
  pthread_t t[_max_thread];
  for(int i = 0; i < _max_thread; ++i)
  {
    pthread_create(t + i, nullptr, Runtine, this);
  }
}

注意： 线程创建后，执行启动函数，在这个函数中，线程会去任务队列中取任务并处理，取任务前需要进行加锁的操作（如果队列为空需要挂起等待），取完任务然后进行解锁，然后处理任务，让其它线程去任务队列中取任务

放任务： 主线程无脑往任务队列中塞任务，塞任务之前进行加锁，塞完任务解锁，然后唤醒在条件变量下等待的队列 取任务： 线程池中的线程从任务队列中取任务，这里不需要加锁，因为这个动作在启动函数中加锁的那一段区间中被调用的，其实已经上锁了

// 放任务
void Put(Task* data)
{
  LockQueue();
  _q.push(data);
  UnlockQueue();
  WakeUpThread();
}
// 取任务
void Get(Task*& data)
{
  data = _q.front();
  _q.pop();
}

这两个版本都可以实现简易的线程池，下面线程池版本的服务器主要是用版本二来实现，因为版本一要修改的内容有点多，小伙伴们可以自己修改一下

线程池版本服务器

多线程版本效果看起来还不错，但是来一个连接就创建一个线程，断开一个连接就释放一个线程，这样频繁地创建和释放线程资源，对OS来说是一种负担，同时也带来资源的浪费，如果我们使用线程池，把每一个客户端连接封装成一个任务，让线程池去处理，这样就不需要频繁地创建和销毁消除，效率也能提升很多。 线程池采用版本二，代码如下：

#pragma once
#include <iostream>
#include <queue>
#include <pthread.h>
#include <unistd.h>
#include "Task.hpp"

#define DEFAULT_MAX_PTHREAD 5

class ThreadPool
{
public:
  ThreadPool(int max_pthread = DEFAULT_MAX_PTHREAD)
    :_max_thread(max_pthread)
  {}
  static void* Runtine(void* arg)
  {
    pthread_detach(pthread_self());
    ThreadPool* this_p = (ThreadPool*)arg;

    while (1){
      this_p->LockQueue();
      while (this_p->IsEmpty()){
        this_p->ThreadWait();
      }
      Task* t;
      this_p->Get(t);
      this_p->UnlockQueue();
      // 解锁后处理任务
      t->Run();
      delete t;
    }
  }
  void ThreadPoolInit()
  {
    pthread_mutex_init(&_mutex, nullptr);
    pthread_cond_init(&_cond, nullptr);
    pthread_t t[_max_thread];
    for(int i = 0; i < _max_thread; ++i)
    {
      pthread_create(t + i, nullptr, Runtine, this);
    }
  }
  void Put(Task* data)
  {
    LockQueue();
    _q.push(data);
    UnlockQueue();
    WakeUpThread();
  }
  void Get(Task*& data)
  {
    data = _q.front();
    _q.pop();
  }
  ~ThreadPool()
  {
    pthread_mutex_destroy(&_mutex);
    pthread_cond_destroy(&_cond);
  }
public:
  void LockQueue()
  {
    pthread_mutex_lock(&_mutex);
  }
  void UnlockQueue()
  {
    pthread_mutex_unlock(&_mutex);
  }
  void ThreadWait()
  {
    pthread_cond_wait(&_cond, &_mutex);
  }
  void WakeUpThread()
  {
    pthread_cond_signal(&_cond);
    //pthread_cond_broadcast(&_cond);
  }
  bool IsEmpty()
  {
    return _q.empty();
  }
private:
  std::queue<Task*>  _q;
  int             _max_thread;
  pthread_mutex_t _mutex;
  pthread_cond_t  _cond;
};

这里我们单独写一个头文件——Task.hpp，其中有任务类，任务类里面有三个成员变量，也就是端口号，IP和套接字，其中有一个成员方法——Run，里面封装了一个Service函数，也就是前面写的，把它放在Task.hpp这个头文件下，线程池里面的线程执行run函数即可，头文件内容如下：

#pragma once
#include <iostream>
#include <unistd.h>

static void Service(std::string ip, int port, int sock)
{
  while (1){
    char buf[256];
    ssize_t size = read(sock, buf, sizeof(buf)-1);
    if (size > 0){
      // 正常读取size字节的数据
      buf[size] = 0;
      std::cout << "[" << ip << "]:[" << port  << "]# "<< buf << std::endl;
      std::string msg = "server get!-> ";
      msg += buf;
      write(sock, msg.c_str(), msg.size());
    }
    else if (size == 0){
      // 对端关闭
      std::cout << "[" << ip << "]:[" << port  << "]# close" << std::endl;
      break;
    }
    else{
      // 出错
      std::cerr << sock << "read error" << std::endl;
      break;
    }
  }

  close(sock);
  std::cout << "service done" << std::endl;
}

struct Task
{
  int _port;
  std::string _ip;
  int _sock;

  Task(int port, std::string ip, int sock)
    :_port(port)
    ,_ip(ip)
     ,_sock(sock)
  {}
  void Run()
  {
      Service(_ip, _port, _sock);
  }
};

服务器类的核心代码如下：

void loop()
{
  struct sockaddr_in peer;// 获取远端端口号和ip信息
  socklen_t len = sizeof(peer);
  _tp = new ThreadPool(THREAD_NUM);
  _tp->ThreadPoolInit();
  while (1){
    // 获取链接
    // sock 是进行通信的一个套接字  _listen_sock 是进行监听获取链接的一个套接字
    int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
    if (sock < 0){
      std::cout << "accept fail, continue accept" << std::endl;
      continue;
    }
    int peerPort = ntohs(peer.sin_port);
    std::string peerIp = inet_ntoa(peer.sin_addr);
    std::cout << "get a new link, [" << peerIp << "]:[" << peerPort  << "]"<< std::endl;
    Task* task = new Task(peerPort, peerIp, sock);
    _tp->Put(task);

  }
}

注意几点变化：

服务器类增加一个线程池成员变量，初始化函数里面增加线程池创建（在堆上申请）
析构函数增加释放线程池资源一步
loop函数中只需要封装任务，并把任务丢进线程池中即可

标签：