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。

相关文章

C++右移运算符的一个小坑及解决

C++右移运算符的一个小坑及解决

《C++右移运算符的一个小坑及解决》文章指出右移运算符处理负数时左侧补1导致死循环,与除法行为不同,强调需注意补码机制以正确统计二进制1的个数... 目录我遇到了这么一个www.chinasem.cn函数由此可以看到也很好理解总结我遇到了这么一个函数template<typename T>unsigned
阅读更多...
Python实现批量CSV转Excel的高性能处理方案

Python实现批量CSV转Excel的高性能处理方案

《Python实现批量CSV转Excel的高性能处理方案》在日常办公中,我们经常需要将CSV格式的数据转换为Excel文件,本文将介绍一个基于Python的高性能解决方案,感兴趣的小伙伴可以跟随小编一... 目录一、场景需求二、技术方案三、核心代码四、批量处理方案五、性能优化六、使用示例完整代码七、小结一、
阅读更多...
C++统计函数执行时间的最佳实践

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法
阅读更多...
Python中logging模块用法示例总结

Python中logging模块用法示例总结

《Python中logging模块用法示例总结》在Python中logging模块是一个强大的日志记录工具,它允许用户将程序运行期间产生的日志信息输出到控制台或者写入到文件中,：本文主要介绍Pyt... 目录前言一. 基本使用1. 五种日志等级2.  设置报告等级3. 自定义格式4. C语言风格的格式化方法
阅读更多...
Web服务器-Nginx-高并发问题

Web服务器-Nginx-高并发问题

《Web服务器-Nginx-高并发问题》Nginx通过事件驱动、I/O多路复用和异步非阻塞技术高效处理高并发,结合动静分离和限流策略,提升性能与稳定性... 目录前言一、架构1. 原生多进程架构2. 事件驱动模型3. IO多路复用4. 异步非阻塞 I/O5. Nginx高并发配置实战二、动静分离1. 职责2
阅读更多...
C# LiteDB处理时间序列数据的高性能解决方案

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章：LiteDB时间序列数据模型设计1.1 核心设计原则
阅读更多...
Python 基于http.server模块实现简单http服务的代码举例

Python 基于http.server模块实现简单http服务的代码举例

《Python基于http.server模块实现简单http服务的代码举例》Pythonhttp.server模块通过继承BaseHTTPRequestHandler处理HTTP请求,使用Threa... 目录测试环境代码实现相关介绍模块简介类及相关函数简介参考链接测试环境win11专业版python
阅读更多...
Android协程高级用法大全

Android协程高级用法大全

《Android协程高级用法大全》这篇文章给大家介绍Android协程高级用法大全,本文结合实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友跟随小编一起学习吧... 目录1️⃣ 协程作用域（CoroutineScope）与生命周期绑定Activity/Fragment 中手
阅读更多...
深入解析C++ 中std::map内存管理

深入解析C++ 中std::map内存管理

《深入解析C++中std::map内存管理》文章详解C++std::map内存管理,指出clear()仅删除元素可能不释放底层内存,建议用swap()与空map交换以彻底释放,针对指针类型需手动de... 目录1️、基本清空std::map2️、使用 swap 彻底释放内存3️、map 中存储指针类型的对象
阅读更多...
Linux搭建ftp服务器的步骤

Linux搭建ftp服务器的步骤

《Linux搭建ftp服务器的步骤》本文给大家分享Linux搭建ftp服务器的步骤,本文通过图文并茂的形式给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录ftp搭建1：下载vsftpd工具2：下载客户端工具3：进入配置文件目录vsftpd.conf配置文件4:
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2