482-手写C++muduo库(Thread,EventLoopThread,EventLoopThreadPool)

深藏阁楼爱情的钟 2022-09-08 13:59 129阅读 0赞

**EventLoopThreadPool：事件循环的线程池，管理事件循环调度**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]  
**我们点进去EventLoopThreadPool，看到**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
**打包了一个eventloop和一个线程**  
**我们再进去Thread里面**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
**就是底层的Thread，我们用C++11精简**

**我们分别实现上面的3个类，从小到大输出**

## Thread线程类 ##

**Thread.h**

#pragma once
    
    #include "noncopyable.h"
    
    #include <functional>
    #include <thread>
    #include <memory>
    #include <unistd.h>
    #include <string>
    #include <atomic>
    
    class Thread : noncopyable
    {
        
    public:
        using ThreadFunc = std::function<void()>;//线程函数的函数类型  绑定器和函数对象，就可以传参 
    
        explicit Thread(ThreadFunc, const std::string &name = std::string());//构造函数 
        ~Thread();//析构函数 
    
        void start();//启动当前线程 
        void join();//当前线程等待其他线程完了再运行下去 
    
        bool started() const {
         return started_; }
        pid_t tid() const {
         return tid_; }
        const std::string& name() const {
         return name_; }
    
        static int numCreated() {
         return numCreated_; }
    private:
        void setDefaultName();
    
        bool started_;//启动当前线程 
        bool joined_;//当前线程等待其他线程完了再运行下去 
        std::shared_ptr<std::thread> thread_;//自己来掌控线程对象产生的时机 
        pid_t tid_;
        ThreadFunc func_;//存储线程函数 
        std::string name_;//调试的时候打印 
        static std::atomic_int numCreated_;//对线程数量计数 
    };

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
**我们用C++结合lambda表达式非常方便**  
**Thread.cc**

#include "Thread.h"
    #include "CurrentThread.h"
    
    #include <semaphore.h>
    
    std::atomic_int Thread::numCreated_(0);//初始化
    
    Thread::Thread(ThreadFunc func, const std::string &name)//构造函数
        : started_(false)
        , joined_(false)
        , tid_(0)
        , func_(std::move(func))
        , name_(name)
    {
        
        setDefaultName();
    }
    
    Thread::~Thread()//析构函数
    {
        
        if (started_ && !joined_)//线程已经运行起来并且不是工作线程join
        {
        
            thread_->detach();
    		//thread类提供的设置分离线程的方法，成了1个守护线程，当主线程结束，守护线程自动结束
        }
    }
    
    void Thread::start()//一个Thread对象，记录的就是一个新线程的详细信息
    {
        
        started_ = true;
        sem_t sem;
        sem_init(&sem, false, 0);
    
        //开启线程
        thread_ = std::shared_ptr<std::thread>(new std::thread([&](){
        
            //获取线程的tid值
            tid_ = CurrentThread::tid();
            sem_post(&sem);//加1操作
            //开启一个新线程，专门执行该线程函数
            func_();//包含一个eventloop
        }));//智能指针指向线程对象
    
        //这里必须等待获取上面新创建的线程的tid值
        sem_wait(&sem);//前面加1之后才能获取，才能解除阻塞
    }
    
    void Thread::join()
    {
        
        joined_ = true;
        thread_->join();
    }
    
    void Thread::setDefaultName()//给线程设置默认的名字
    {
        
        int num = ++numCreated_;
        if (name_.empty())//线程还没有名字
        {
        
            char buf[32] = {
        0};
            snprintf(buf, sizeof buf, "Thread%d", num);
            name_ = buf;
        }
    }

## EventLoopThread ##

**绑定了一个loop和thread，在一个thread创建一个loop**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
**EventLoopThread.h**

#pragma once
    
    #include "noncopyable.h"
    #include "Thread.h"
    
    #include <functional>
    #include <mutex>
    #include <condition_variable>
    #include <string>
    
    class EventLoop;
    
    class EventLoopThread : noncopyable
    {
        
    public:
        using ThreadInitCallback = std::function<void(EventLoop*)>; 
    
        EventLoopThread(const ThreadInitCallback &cb = ThreadInitCallback(), //线程初始化的回调 
            const std::string &name = std::string());
        ~EventLoopThread();
    
        EventLoop* startLoop();//开启循环 
    private:
        void threadFunc();//线程函数，创建loop 
    
        EventLoop *loop_;
        bool exiting_;//是否退出循环 
        Thread thread_;
        std::mutex mutex_;
        std::condition_variable cond_;
        ThreadInitCallback callback_;//初始化操作 
    };

**EventLoopThread.cc**

#include "EventLoopThread.h"
    #include "EventLoop.h"
    
    
    EventLoopThread::EventLoopThread(const ThreadInitCallback &cb, 
            const std::string &name)
            : loop_(nullptr)
            , exiting_(false)
            , thread_(std::bind(&EventLoopThread::threadFunc, this), name)//绑定回调函数
            , mutex_()
            , cond_()
            , callback_(cb)
    {
        
    
    }
    
    EventLoopThread::~EventLoopThread()//析构函数
    {
        
        exiting_ = true;
        if (loop_ != nullptr)
        {
        
            loop_->quit();
            thread_.join();
        }
    }
    
    EventLoop* EventLoopThread::startLoop()//开启循环
    {
        
        thread_.start();//启动底层的新线程
    	//启动后执行的是EventLoopThread::threadFunc
        EventLoop *loop = nullptr;
        {
        
            std::unique_lock<std::mutex> lock(mutex_);
            while ( loop_ == nullptr )
            {
        
                cond_.wait(lock);//挂起，等待
            }
            loop = loop_;
        }
        return loop;
    }
    
    //下面这个方法，start（）后的执行的，也就是在单独的新线程里面运行的
    void EventLoopThread::threadFunc()
    {
        
        EventLoop loop;//创建一个独立的eventloop，和上面的线程是一一对应的，one loop per thread
    
        if (callback_)//如果有回调
        {
        
            callback_(&loop);//绑定loop做一些事情
        }
    
        {
        
            std::unique_lock<std::mutex> lock(mutex_);
            loop_ = &loop;//就是运行在这个线程的loop对象
            cond_.notify_one();//唤醒1个线程
        }
    
        loop.loop();//相当于EventLoop loop  => Poller.poll
        std::unique_lock<std::mutex> lock(mutex_);
        loop_ = nullptr;
    }

## EventLoopThreadPool ##

这个很明显，是池的概念。是一个事件线程池，管理eventloop，eventloop绑定的就是一个线程。

**EventLoopThreadPool.h**

#pragma once
    #include "noncopyable.h"
    
    #include <functional>
    #include <string>
    #include <vector>
    #include <memory>
    
    class EventLoop;
    class EventLoopThread;
    
    class EventLoopThreadPool : noncopyable
    {
        
    public:
        using ThreadInitCallback = std::function<void(EventLoop*)>; 
    
        EventLoopThreadPool(EventLoop *baseLoop, const std::string &nameArg);
        ~EventLoopThreadPool();
    
        void setThreadNum(int numThreads) {
         numThreads_ = numThreads; }//设置底层线程的数量 
    
        void start(const ThreadInitCallback &cb = ThreadInitCallback());//开启整个事件循环线程 
    
        //如果工作在多线程中，baseLoop_默认以轮询的方式分配channel给subloop
        EventLoop* getNextLoop();
    
        std::vector<EventLoop*> getAllLoops();//返回池里的所有loop 
    
        bool started() const {
         return started_; }
        const std::string name() const {
         return name_; }
    private:
    
        EventLoop *baseLoop_;//最基本的loop,
    	//对应一个线程，就是当前用户使用线程 EventLoop loop;负责用户的连接，已连接用户的读写 
    	
        std::string name_;
        bool started_;
        int numThreads_;
        int next_;
        std::vector<std::unique_ptr<EventLoopThread>> threads_;//所有事件的线程 
        std::vector<EventLoop*> loops_;//事件线程的eventloop指针 
    };

**EventLoopThreadPool.cc**

#include "EventLoopThreadPool.h"
    #include "EventLoopThread.h"
    
    #include <memory>
    
    EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseLoop, const std::string &nameArg)
        : baseLoop_(baseLoop)
        , name_(nameArg)
        , started_(false)
        , numThreads_(0)
        , next_(0)
    {
        }
    
    EventLoopThreadPool::~EventLoopThreadPool()
    {
        }
    
    void EventLoopThreadPool::start(const ThreadInitCallback &cb)
    {
        
        started_ = true;
    
        for (int i = 0; i < numThreads_; ++i)
        {
        
            char buf[name_.size() + 32];
            snprintf(buf, sizeof buf, "%s%d", name_.c_str(), i);
            EventLoopThread *t = new EventLoopThread(cb, buf);
            threads_.push_back(std::unique_ptr<EventLoopThread>(t));//不想手动delete
            loops_.push_back(t->startLoop());//底层创建线程，绑定一个新的EventLoop，并返回该loop的地址
        }
    
        //整个服务端只有一个线程，运行着baseloop，就是用户创建的mainloop
        if (numThreads_ == 0 && cb)
        {
        
            cb(baseLoop_);
        }
    }
    
    //如果工作在多线程中，baseLoop_默认以轮询的方式分配channel给subloop
    EventLoop* EventLoopThreadPool::getNextLoop()
    {
        
        EventLoop *loop = baseLoop_;//用户创建的mainloop
    
        if (!loops_.empty())//通过轮询获取下一个处理事件的loop
        {
        
            loop = loops_[next_];
            ++next_;
            if (next_ >= loops_.size())
            {
        
                next_ = 0;
            }
        }
    
        return loop;
    }
    
    std::vector<EventLoop*> EventLoopThreadPool::getAllLoops()
    {
        
        if (loops_.empty())
        {
        
            return std::vector<EventLoop*>(1, baseLoop_);
        }
        else
        {
        
            loops_;
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/2c8dcb77f138425180267bc61b6c3420.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/45b1fe3fd1f74575b798c120dff2a42a.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/6b6b0c564e9f46d7abdb51224168c985.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/00a8b57f77be483184c124c13786f585.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/cdd0739149574b19831bd07fde92183d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/82d9f97d0f2c46aa8c637f9c07827451.png