Sylar C++高性能服务器学习记录09 【协程调度模块-知识储备篇】

2024-05-08 12:36
文章标签 c++ 模块 学习 服务器 记录 知识 高性能 调度 09 协程 sylar 储备

本文主要是介绍Sylar C++高性能服务器学习记录09 【协程调度模块-知识储备篇】，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

早在19年5月就在某站上看到sylar的视频了，一直认为这是一个非常不错的视频，由于本人一直是自学编程，基础不扎实，也没有任何人的督促，没能坚持下去，每每想起倍感惋惜。恰逢互联网寒冬，在家无事，遂提笔再续前缘。

为了能更好的看懂sylar，本套笔记会分两步走，每个系统都会分为两篇博客。
分别是【知识储备篇】和【代码分析篇】
(ps:纯粹做笔记的形式给自己记录下，欢迎大家评论，不足之处请多多赐教)
QQ交流群：957100923

协程调度模块-代码分析篇

一、Schedule调度器是什么【线程级别的调度器】

我的博客已经停更一周了，并不是 五一 假期出去游玩导致的。
原因很简单：就是因为 sylar 视频中协程调度相关的知识过于复杂，导致自己反复观看了数遍才勉强理解一二。
要搞懂协程调度模块一上来就看sylar的视频或代码未免门槛太高了。
我总结了一下这部分知识为什么难以下咽：

  1. 知识点过多，将调度器、线程、协程糅合在一起讲解，原本基础就不扎实，这样一锅乱炖容易糊掉。
  2. 想法没有提前告知，不知道要做什么，在做什么，雾里看花。
  3. 弹幕的误导，sylar-yin 确实会有说错写错的地方，如果你不懂，那么请你百分百遵循他的写法。
  4. 不够坚持，不要放过每一个你不熟悉的知识点，请暂停视频，把不会的知识点了解了再往下(不要过分深入)。

本人属于不太聪明的类型，所以如果你和我一样，那么请稳扎稳打，别人学一天我们就学三天，别人学一个月，我们就学一年。

1.最普通的多线程使用

首先抛开协程，我们想要使用多线程来完成多个任务需要怎么做呢？
下面我举个例子(该案例中会开启三个子线程来分别执行 唱歌、跳舞、说唱)：

#include <chrono>
#include <iostream>
#include <pthread.h>
#include <vector>
#include <list>
#include <unistd.h>
#include <functional>
#include "sylar/thread.h"
#include "sylar/mutex.h"//测试用的全局累加计数字段
int num = 0;
//循环累加的次数
uint64_t loop_times = 100000000;typedef sylar::Mutex MutexType;
MutexType g_mutx;//唱歌的方法，会专门分配一个线程来处理
void sing(){MutexType::Lock lock(g_mutx);for(size_t i = 0; i < loop_times; ++i){++num;}std::cout << sylar::Thread::GetName() << " sing~ " << num << std::endl;
}//跳舞的方法，会专门分配一个线程来处理
void dance(){MutexType::Lock lock(g_mutx);for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " dance~ " << num << std::endl;   
}//说唱的方法，会专门分配一个线程来处理
void rap(){MutexType::Lock lock(g_mutx);for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " rap~ " << num << std::endl;   
}int main(){std::cout << "====Main start====" << std::endl;{//记录当前时间auto start = std::chrono::high_resolution_clock::now();//创建线程用于执行唱歌的方法sylar::Thread::ptr thr1(new sylar::Thread(&sing,"THREAD_1"));//创建线程用于执行跳舞的方法sylar::Thread::ptr thr2(new sylar::Thread(&dance,"THREAD_2"));//创建线程用于执行说唱的方法sylar::Thread::ptr thr3(new sylar::Thread(&rap,"THREAD_3"));thr1->join();thr2->join();thr3->join();//记录结束时间auto end = std::chrono::high_resolution_clock::now();//计算花费的时间std::chrono::duration<double,std::milli> elapsed = end - start;std::cout << "Elapsed: " << (int)elapsed.count() << std::endl;}std::cout << "====Main end====" << std::endl;return 0;
}

以下是输出(可以看到三个线程按顺序执行了):

===Main start===
THREAD_1 sing~ 100000000
THREAD_2 dance~ 200000000
THREAD_3 rap~ 300000000
Elapsed: 294
===Main end====
2.复用多线程

上述的代码中，我们有三个任务（唱歌、跳舞、说唱），同时我们开辟了三个线程来执行。
但是很多情况下我们的线程数量是需要控制在适当范围内的。

最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）* CPU数目CPU核数为4核，一个任务线程cpu耗时为20ms，线程等待（网络IO、磁盘IO）耗时80ms，
那最佳线程数目：( 80 + 20 )/20 * 4 = 20。也就是设置20个线程数最佳。线程的等待时间越大，线程数就要设置越大
1、CPU密集型：操作内存处理的业务，一般线程数设置为：CPU核数 + 1 或者 CPU核数*2。核数为4的话，一般设置 5 或 8
2、IO密集型：文件操作，网络操作，数据库操作，一般线程设置为：cpu核数 / (1-0.9)，核数为4的话，一般设置 40

我的机器是1核的，我这里其实适合使用两个线程，但是我在上述代码中使用了三个线程，那么多线程反而变成了低效的。
所以我们可以复用其中一个线程，在该线程执行完成任务后再执行其他的任务。
代码如下（改动很少，改动部分用 !!!标注了）：

int main(){std::cout << "====Main start====" << std::endl;{//记录当前时间auto start = std::chrono::high_resolution_clock::now();//创建线程用于执行唱歌的方法sylar::Thread::ptr thr1(new sylar::Thread(&sing,"THREAD_1"));//创建线程用于执行跳舞的方法sylar::Thread::ptr thr2(new sylar::Thread(&dance,"THREAD_2"));thr1->join();thr2->join();//!!!复用线程用于执行说唱的方法!!!thr1.reset(new sylar::Thread(&rap,"THREAD_1"));thr1->join();//记录结束时间auto end = std::chrono::high_resolution_clock::now();//计算花费的时间std::chrono::duration<double,std::milli> elapsed = end - start;std::cout << "Elapsed: " << (int)elapsed.count() << std::endl;}std::cout << "====Main end====" << std::endl;return 0;
}

以下是输出：

===Main start====
THREAD_1 sing~ 100000000
THREAD_2 dance~ 200000000
THREAD_1 rap~ 300000000
Elapsed: 265
==Main end====

问题：
问题来了，以上的两份代码都存在以下问题：
1.就是需要程序员自己来控制线程的创建。
2.需要程序员手动来进行线程的调度。
3.需要程序员手动的去复用一个线程对象。
4.需要程序员手动的join对应的每一个线程。
这样是很繁琐的事情，而且一不小心就会出错，那么我们有什么好的办法吗？

3.调度器的出现

想要解决以上代码带来的问题，我们就要提出需求：
1.将创建线程的事情交由【调度器】执行，我们只需创建调度器时指定线程数量即可。
2.将线程调度和复用线程的事情交由【调度器】执行，我们只需提供所需执行的 任务 即可。
3.等待各个线程完成的jion()方法也由【调度器】执行，我们只需等待调度器结束即可。

综上所述，我们给出以下代码(记得先看main方法中的调度器使用，再看实现)：

#include <chrono>
#include <iostream>
#include <pthread.h>
#include <vector>
#include <list>
#include <unistd.h>
#include <functional>
#include "sylar/thread.h"
#include "sylar/mutex.h"//测试用的全局累加计数字段
int num = 0;
//循环累加的次数
uint64_t loop_times = 100000000;//唱歌的方法，会专门分配一个线程来处理
void sing(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout << sylar::Thread::GetName() << " sing~ " << num << std::endl;
}//跳舞的方法，会专门分配一个线程来处理
void dance(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " dance~ " << num << std::endl;   
}//说唱的方法，会专门分配一个线程来处理
void rap(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " rap~ " << num << std::endl;   
}//调度器
class Schedule {
public://定义锁，使用typedef方便锁类型替换typedef sylar::Mutex MutexType;//调度器构造函数，指定线程池的大小Schedule(size_t size = 1):m_poll_size(size){std::cout << "Specify a thread poll size of "<< size << std::endl;}//调度器析构函数~Schedule(){}//调度方法void run(){//多个线程操作任务队列，涉及到线程安全，所以要加锁MutexType::Lock lock(m_mutex);//定义一个任务对象来存放后续从任务队列中取出的任务std::function<void()> task;//开始对任务队列进行迭代auto it = m_tasks_queue.begin();while(it != m_tasks_queue.end()){//将任务从任务队列中取出task = *it;//将取出的任务从任务队列中擦除m_tasks_queue.erase(it);//每次按顺序取出一个，取出后跳出循环break;}//判断是否有取到任务if(task != nullptr){//有取到任务则执行任务task();}}//开启调度器方法void start(){std::cout << "Schedule start..." << std::endl;std::cout << "Init a thread poll" << std::endl;//根据调度器初始化时指定的线程池大小，来初始化线程池容器的大小 m_thread_poll.resize(m_poll_size);std::cout << "Schedule run..." << std::endl;//构建线程池中的每一个线程对象，每个线程都将绑定调度器的run方法，且给每个线程对象指定了名称用于区分for(size_t i = 0; i < m_poll_size; ++i){//这里的reset方法是为了复用线程对象节约内存资源m_thread_poll[i].reset(new sylar::Thread(std::bind(&Schedule::run,this),"THREAD_"+std::to_string(i)));}}//停止调度器方法void stop(){//判断任务队列中的任务是否全部被消化while(!m_tasks_queue.empty()){for(size_t i = 0; i < m_poll_size; ++i){m_thread_poll[i].reset(new sylar::Thread(std::bind(&Schedule::run,this),"THREAD_"+std::to_string(i)));}}//如果任务队列中的任务全部被消化了，那么就在这等待每个子线程的结束for(size_t i = 0; i < m_poll_size; ++i){m_thread_poll[i]->join();}std::cout << "Schedule stop..." << std::endl; }//接受调度任务方法void schedule(std::function<void()> task){std::cout << "Push task into queue " << &task << std::endl;//将任务对象依次存放到调度器的任务队列中m_tasks_queue.push_back(task);}private://由于是多线程环境下所以提供锁MutexType m_mutex;//线程池的大小size_t m_poll_size = 0;//线程池std::vector<sylar::Thread::ptr> m_thread_poll;//任务队列std::list<std::function<void()> > m_tasks_queue;      
};int main(){std::cout << "====Schedule====" << std::endl;std::cout << "====Main start====" << std::endl;{//记录当前时间auto start = std::chrono::high_resolution_clock::now();//构建调度器，指定线程池中线程数量Schedule sc(2);//依次将要执行的任务塞入调度器sc.schedule(std::bind(&sing));sc.schedule(std::bind(&dance));sc.schedule(std::bind(&rap));//执行调度器sc.start();//停止调度器sc.stop();//记录结束时间auto end = std::chrono::high_resolution_clock::now();//计算花费的时间std::chrono::duration<double,std::milli> elapsed = end - start;std::cout << "Elapsed: " << (int)elapsed.count() << std::endl;}std::cout << "====Main end====" << std::endl;return 0;
}

以下是输出：

===Schedule===
===Main start====
Push task into queue  0x7ffd602ffb00
Push task into queue  0x7ffd602ffb40
Push task into queue  0x7ffd602ffb80
Schedule start...
Init a thread poll
Schedule run...
THREAD_1 sing~ 100000000
THREAD_2 dance~ 200000000
THREAD_1 rap~ 300000000
Schedule stop...
Elapsed: 234
===Main end====

可以看到以上的代码使用了调度器，代码的编写会很舒服，以下是使用调度器代码的部分：

//构建调度器，指定线程池中线程数量
Schedule sc(2);
//依次将要执行的任务塞入调度器
sc.schedule(std::bind(&sing));
sc.schedule(std::bind(&dance));
sc.schedule(std::bind(&rap));
//执行调度器
sc.start();
//停止调度器
sc.stop();

这样一来，无论是有多少任务需要执行，程序员都不需要关系如何创建线程和如何进行调度了，
只需要把所要执行的任务全部一股脑的交给调度器就行了。
线程也不需要自己来维护了，只需要告诉调度器需要多少个线程就行。
只要调度器完善的够好，程序员就只需要专心处理业务问题就好了。

看到这里相信你对调度器的功能和必要性有了一定的了解，那么我们来继续优化这个调度器，让它更健硕。

二、考虑多线程下的多个调度器问题【线程级别的调度器】

  1. 由于上边代码调度器Schedule类可以在其他任意线程中构建与执行，所以需要考虑多线程下调度器的区分问题。
  2. 由于这里为了由浅入深，所以没有使用协程，那么调度器就必须占用一个线程来执行。

接下来的场景会变成：一个调度器线程和多个其他的线程和一系列的函数任务。
所以需要使用线程局部变量来控制一个线程只存在一个调度器
当然这个判断需要将调度器的this存储起来做比较

以下是控制一个线程只存在一个调度器的核心逻辑

static thread_local Schedule* t_schedule = nullptr;Schedule(size_t size = 1):m_poll_size(size){//断言当前线程是否不存在调度器，如果已存在那么报错SYLAR_ASSERT2(GetThis()==nullptr,"one thread one schedule");//设置当前线程上的调度器为当前调度器setThis();
}~Schedule(){//如果当前调度器就是对应线程上的调度器那么将线程上的调度器清空if(GetThis()==this){t_schedule = nullptr;}
}static Schedule* GetThis(){return t_schedule;
}void setThis(){t_schedule = this;
}

例子（完整代码）：

#include <chrono>
#include <iostream>
#include <pthread.h>
#include <vector>
#include <list>
#include <unistd.h>
#include <functional>
#include "sylar/log.h"
#include "sylar/thread.h"
#include "sylar/mutex.h"
#include "sylar/macro.h"int num = 0;
uint64_t loop_times = 100000000;void sing(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout << sylar::Thread::GetName() << " sing~ " << num << std::endl;
}void dance(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " dance~ " << num << std::endl;   
}void rap(){for(size_t i = 0; i < loop_times; ++i){++num;}std::cout <<sylar::Thread::GetName() << " rap~ " << num << std::endl;   
}class Schedule;static thread_local Schedule* t_schedule = nullptr;class Schedule {
public:typedef sylar::Mutex MutexType;Schedule(size_t size = 1):m_poll_size(size){SYLAR_ASSERT2(GetThis()==nullptr,"one thread one schedule");setThis();m_callerThread = sylar::Thread::GetThis();}~Schedule(){if(GetThis()==this){t_schedule = nullptr;}}static Schedule* GetThis(){return t_schedule;}void setThis(){t_schedule = this;}void run(){MutexType::Lock lock(m_mutex);std::function<void()> task;auto it = m_tasks_queue.begin();while(it != m_tasks_queue.end()){task = *it;m_tasks_queue.erase(it);break;}if(task != nullptr){task();}}void start(){std::cout << "Schedule start..." << std::endl;std::cout << "Init a thread poll" << std::endl;m_thread_poll.resize(m_poll_size);std::cout << "Schedule run..." << std::endl;for(size_t i = 0; i < m_poll_size; ++i){m_thread_poll[i].reset(new sylar::Thread(std::bind(&Schedule::run,this),"THREAD_"+std::to_string(i)));}}void stop(){while(!m_tasks_queue.empty()){for(size_t i = 0; i < m_poll_size; ++i){m_thread_poll[i].reset(new sylar::Thread(std::bind(&Schedule::run,this),"THREAD_"+std::to_string(i)));}}for(size_t i = 0; i < m_poll_size; ++i){m_thread_poll[i]->join();}std::cout << "Schedule stop..." << std::endl; }void schedule(std::function<void()> task){std::cout << "Push task into queue " << &task << std::endl;m_tasks_queue.push_back(task);}private:MutexType m_mutex;size_t m_poll_size = 0;std::vector<sylar::Thread::ptr> m_thread_poll;std::list<std::function<void()> > m_tasks_queue;      sylar::Thread* m_callerThread = 0;
};//第二个调度线程
void subThread(){std::cout << "Sub Schedule Thread" << std::endl;Schedule sc(1);sc.schedule(std::bind(&rap));sc.start();sc.stop();
}int main(){std::cout << "====Schedule====" << std::endl;std::cout << "====Main start====" << std::endl;{auto start = std::chrono::high_resolution_clock::now();Schedule sc(2);sc.schedule(std::bind(&subThread));sc.schedule(std::bind(&dance));// sc.schedule(std::bind(&rap));sc.start();sc.stop();auto end = std::chrono::high_resolution_clock::now();std::chrono::duration<double,std::milli> elapsed = end - start;std::cout << "Elapsed: " << (int)elapsed.count() << std::endl;}std::cout << "====Main end====" << std::endl;return 0;
}

可以看到，调度器可以在任何一个线程中使用，互不影响。

三、加入协程的概念【协程级别的调度器】

一旦加入协程的概念，那么调度器的复杂度就会上升一个层次。
首先要理清楚有哪些东西参与到这个调度器中。
其次要统一称呼，视频中就是很多称呼不明确导致看不懂。
需要理清楚以下几个点：

  1. 加入协程的概念后，调度器就不需要占用一个线程来专门做调度，而是退居到某个线程下的 协程 中去。
  2. 一旦退居到某个协程中，那么该调度器所在的协程我们叫做 【调度协程】。
  3. 【调度协程】 所在的线程我称它为 【调度线程】。
  4. 在一个线程中会有一个 【主协程】 ，要记住 【调度协程】 可以是 【主协程】 也可能不是(调度线程是主线程时必然不是)。
  5. 在多线程中会有一个 【主线程】，要记住 【调度线程】 可以是 【主线程】 也可以不是。
  6. 调度任务会变成两种类型：【函数任务】、【协程任务】，这里可以把他们包装到一起，统称 【任务】。
  7. 【函数任务】最终可以包装成【协程任务】，这样统一调用方式。
  8. 我们封装的协程对象会有很多状态，所以在调度器中需要有比较复杂的判断（这个需要看代码了）。

四、总结

1.要看懂协程调度必须拆分着看，辩证的看。
2.由于协程调度的IO协程篇还没看完，所以存在几个遗留方法，这样无法看到全貌，可以理解大概流程后继续。
3.作者代码中确实有不足的地方，需要先内心赞同他的一切做法，当自己完全看懂后再做修改(比如Fiber中有Schedule的耦合代码)。
4.本人还没完全看完协程调度相关的视频，在看完IO协程调度后可能会回过头来修改这部分的博客内容。

这篇关于Sylar C++高性能服务器学习记录09 【协程调度模块-知识储备篇】的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/970332。

相关文章

Java深度学习库DJL实现Python的NumPy方式

Java深度学习库DJL实现Python的NumPy方式

《Java深度学习库DJL实现Python的NumPy方式》本文介绍了DJL库的背景和基本功能,包括NDArray的创建、数学运算、数据获取和设置等,同时,还展示了如何使用NDArray进行数据预处理... 目录1 NDArray 的背景介绍1.1 架构2 JavaDJL使用2.1 安装DJL2.2 基本操
阅读更多...
关于Spring @Bean 相同加载顺序不同结果不同的问题记录

关于Spring @Bean 相同加载顺序不同结果不同的问题记录

《关于Spring@Bean相同加载顺序不同结果不同的问题记录》本文主要探讨了在Spring5.1.3.RELEASE版本下,当有两个全注解类定义相同类型的Bean时,由于加载顺序不同,最终生成的... 目录问题说明测试输出1测试输出2@Bean注解的BeanDefiChina编程nition加入时机总结问题说明
阅读更多...
JavaWeb-WebSocket浏览器服务器双向通信方式

JavaWeb-WebSocket浏览器服务器双向通信方式

《JavaWeb-WebSocket浏览器服务器双向通信方式》文章介绍了WebSocket协议的工作原理和应用场景,包括与HTTP的对比,接着,详细介绍了如何在Java中使用WebSocket,包括配... 目录一、概述二、入门2.1 POM依赖2.2 编写配置类2.3 编写WebSocket服务2.4 浏
阅读更多...
c++中std::placeholders的使用方法

c++中std::placeholders的使用方法

《c++中std::placeholders的使用方法》std::placeholders是C++标准库中的一个工具,用于在函数对象绑定时创建占位符,本文就来详细的介绍一下,具有一定的参考价值,感兴... 目录1. 基本概念2. 使用场景3. 示例示例 1：部分参数绑定示例 2：参数重排序4. 注意事项5.
阅读更多...
使用C++将处理后的信号保存为PNG和TIFF格式

使用C++将处理后的信号保存为PNG和TIFF格式

《使用C++将处理后的信号保存为PNG和TIFF格式》在信号处理领域,我们常常需要将处理结果以图像的形式保存下来,方便后续分析和展示,C++提供了多种库来处理图像数据,本文将介绍如何使用stb_ima... 目录1. PNG格式保存使用stb_imagephp_write库1.1 安装和包含库1.2 代码解
阅读更多...
查询SQL Server数据库服务器IP地址的多种有效方法

查询SQL Server数据库服务器IP地址的多种有效方法

《查询SQLServer数据库服务器IP地址的多种有效方法》作为数据库管理员或开发人员,了解如何查询SQLServer数据库服务器的IP地址是一项重要技能,本文将介绍几种简单而有效的方法,帮助你轻松... 目录使用T-SQL查询方法1：使用系统函数方法2：使用系统视图使用SQL Server Configu
阅读更多...
C++实现封装的顺序表的操作与实践

C++实现封装的顺序表的操作与实践

《C++实现封装的顺序表的操作与实践》在程序设计中,顺序表是一种常见的线性数据结构,通常用于存储具有固定顺序的元素,与链表不同,顺序表中的元素是连续存储的,因此访问速度较快,但插入和删除操作的效率可能... 目录一、顺序表的基本概念二、顺序表类的设计1. 顺序表类的成员变量2. 构造函数和析构函数三、顺序表
阅读更多...
使用C++实现单链表的操作与实践

使用C++实现单链表的操作与实践

《使用C++实现单链表的操作与实践》在程序设计中,链表是一种常见的数据结构,特别是在动态数据管理、频繁插入和删除元素的场景中,链表相比于数组,具有更高的灵活性和高效性,尤其是在需要频繁修改数据结构的应... 目录一、单链表的基本概念二、单链表类的设计1. 节点的定义2. 链表的类定义三、单链表的操作实现四、
阅读更多...
Python利用自带模块实现屏幕像素高效操作

Python利用自带模块实现屏幕像素高效操作

《Python利用自带模块实现屏幕像素高效操作》这篇文章主要为大家详细介绍了Python如何利用自带模块实现屏幕像素高效操作,文中的示例代码讲解详,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1、获取屏幕放缩比例2、获取屏幕指定坐标处像素颜色3、一个简单的使用案例4、总结1、获取屏幕放缩比例from
阅读更多...
nginx-rtmp-module模块实现视频点播的示例代码

nginx-rtmp-module模块实现视频点播的示例代码

《nginx-rtmp-module模块实现视频点播的示例代码》本文主要介绍了nginx-rtmp-module模块实现视频点播,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习... 目录预置条件Nginx点播基本配置点播远程文件指定多个播放位置参考预置条件配置点播服务器 192.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2