Muduo网络库解析 ---线程模块

🕗 发布于 2024-12-13 16:29 网络 c++ tcp/ip

前言

注：本文将重点剖析 Muduo 网络库的核心框架，深入探讨作者精妙的代码设计思路，并针对核心代码部分进行重写，将原本依赖 boost 的实现替换为原生的 C++11 语法。需要说明的是，本文并不打算对整个 Muduo 库进行完整的重写。Muduo库源码链接

在前面几篇博客中，我们已经对 基础模块 和 事件循环模块 进行了初步讲解，特别是在事件循环模块中重点介绍了 Channel、Poller 以及 EventLoop 的作用，分析了它们之间的联系以及协同工作的方式。

在此基础上，我们可以进一步思考一个问题：如何保证 EventLoop 在一个线程中独立运行，即如何确保线程对 EventLoop 生命周期的控制？ 用户在使用 Muduo 库时，可以通过 setThreadNum 方法来设置 SubLoop（工作线程）的数量，从而创建多个 EventLoop。这就引出了一个关键点：如何实现并严格保证 One Loop Per Thread 的设计原则？

答案是：我们将线程抽象为 Thread 类，并进一步结合 EventLoop 和 Thread，设计出 EventLoopThread 类。在此基础上，为了实现多个工作线程的管理，引入了线程池的概念，将多个 EventLoopThread 抽象为 EventLoopThreadPool。通过 EventLoopThreadPool，用户只需设置线程数量，即可创建多个事件循环，同时严格遵循了 One Loop Per Thread 的设计原则。

本篇将介绍Muduo网络库的线程有关模块，具体如下：

Thread
EventThread
EventThreadPool

Thread

CurrentThread

在上篇博客中提到：在EventLoop类中有一个成员变量（threadId_ ：const pid_t），其用来标识此EventLoop的线程号。在构造函数中，对其初始化如下：

threadId_(CurrentThread::tid())

CurrentThread::tid(), 现在我们来拨开它神秘的面纱。其定义如下（不是源版，而是精简重写版）：

/*
CurrentThread.h
*/
namespace CurrentThread
{
    /*
        每个线程都有一个唯一的ID
    */
    extern thread_local int t_cachedTid;

    void cacheTid();

    inline int tid()
    {
        if(__builtin_expect(t_cachedTid == 0, 0)){
            cacheTid();
        }
        return t_cachedTid;
    }
}

/*
CurrentThread.cc
*/
namespace CurrentThread
{
    thread_local int t_cachedTid = 0;

    void cacheTid()
    {
        if(t_cachedTid == 0)
        {
            t_cachedTid = static_cast<pid_t>(::syscall(SYS_gettid));
        }
    }
}

解析：

t_cachedTid：
这是一个整型类型，并用thread_local修饰。这意味着每个线程都有一个单独的t_cachedTid，其含义表示为每个线程的tid为CurrentThread::t_cachedTid。每个线程都可以通过CurrentThread::tid()返回它们自己的线程号（因为tid()返回t_cachedTid）。
cacheTid：
此函数的作用为获取本线程的线程号并将其缓存起来。获取TID所用的系统调用为syscall(SYS_gettid)，而pthread_self()获取的也是线程ID。它们两个有什么区别呢？

syscall(SYS_gettid)返回的是当前线程的内核线程 ID（Thread ID，简称 TID）。
pthread_self 获取的是线程库（POSIX Thread Library）中的线程 ID，简称为 pthread_t。
TID是由 Linux 内核分配的，线程在内核中的唯一标识，系统中只有一个TID。

pthread_t是由用户态的线程库管理的，并不是内核线程的唯一标识。
tid()：
其作用为返回当前的线程TID。其中用到了__builtin_expect(t_cachedTid == 0, 0)，表示t_cachedTid == 0这个表达式为假的概率更高。

__builtin_expect 是 GCC 提供的一种编译器内置函数，用于优化分支预测。它的作用是让程序员显式地告诉编译器某个条件的可能性，从而帮助编译器生成更高效的代码。
第一个参数为一个条件表达式，第二个参数表示为表达式哪个（真/假）概率较高。

Thread代码

Thread 类的作用可以概括为：在一个线程中运行指定的函数，同时提供控制该线程（启动、停止等）的接口。Muduo作者对线程进行了封装，在这里我们进行简化利用C++11中的thread库表示。

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();
private:
    static std::atomic_int numCreated_; // 已创建线程的数量

    bool started_;  // 是否开始运行函数
    bool joined_;   // 是否join
    std::shared_ptr<std::thread> thread_;   // 线程
    pid_t tid_;  // TID
    ThreadFunc func_;   // 要运行的函数
    std::string name_;  // 线程的名字
};

Thread.cc

std::atomic_int Thread::numCreated_(0);

Thread::Thread(ThreadFunc func, std::string const &name) :
    started_(false),
    joined_(false),
    tid_(0),
    func_(std::move(func)),
    name_(name)
{
    setDefaultName();
}

Thread::~Thread()
{
    if(started_ && !joined_)
    {
        thread_->detach();
    }
}

void Thread::start()
{
    started_ = true;

    sem_t sem;
    sem_init(&sem, false, 0);     

    // 创建一个线程，执行EventLoopThread::threadFunc
    // 为了获取新线程的线程号，通过信号量机制同步线程
    thread_ = std::shared_ptr<std::thread>(new std::thread([&](){
        tid_ = CurrentThread::tid();

        sem_post(&sem);

        func_();
    }));
    
    sem_wait(&sem);    
}

void Thread::join() 
{
    joined_ = true;
    thread_->join();
}

void Thread::setDefaultName()
{
    int num = ++numCreated_;
    if(!name_.empty())
    {
        char buf[32];
        snprintf(buf, sizeof buf, "Thread%d", num);
        name_ = buf;
    }
}

解析：

start() 方法用于启动一个独立的线程运行 func_，并缓存新创建线程的 TID。为了在 start() 方法运行完成后能够立即获取新创建线程的 TID，需要处理线程间的同步问题。本代码采用了信号量机制（沿用原作者的做法），确保 start() 方法返回时，线程的 TID 已经正确设置，从而可以立即获取该 TID。

EventLoopThread

该类封装了 EventLoop 和 Thread，并将事件循环处理函数作为 Thread 的构造参数，从而确保了 One Loop Per Thread 的设计原则。同时，该类还提供了创建线程并启动事件循环的便捷接口，方便用户高效地管理线程和事件循环的结合。

EventLoopThread.h

class EventLoopThread : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop*)>;

    explicit EventLoopThread(const ThreadInitCallback& cb, 
                            const std::string& name = std::string());
    ~EventLoopThread();

    // 提供给外部开启事件循环的线程
    EventLoop* startLoop();
private:
    // 向Thread传递的函数对象
    void threadFunc();

private:
    EventLoop* loop_;
    bool exiting_;
    Thread thread_;
    std::mutex mutex_;
    std::condition_variable cond_;
    ThreadInitCallback Callback_;
};

解析：

成员变量

loop_：管理的线程Thread内的SubLoop
exiting_：是否正在退出
thread_：管理的线程
cond_：条件变量，确保thread_内的EventLoop的地址能够被EventLoopThread获取

EventLoopThread.cc

EventLoopThread::EventLoopThread(ThreadInitCallback const &cb,
                                 std::string const &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();
    }
}

/*
此函数作用：
    1. 创建一个新线程
    2. 在新线程中定义一个局部变量EventLoop（栈上）
    3. EventLoop开启循环
    4. 返回EventLoop地址
*/
EventLoop *EventLoopThread::startLoop()
{
    // 在EventLoopThread::EventLoopThread中为成员变量thread_绑定了函数对象EventLoopThread::threadFunc，但未创建新线程。
    // 在start中，创建了一个新的线程，并得到新线程的线程号
    thread_.start();

    EventLoop* loop = nullptr;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        while( loop_ == nullptr)
        {
            cond_.wait(lock);
        }
        loop = loop_;
    }
    return loop;
}

// Thread::start()中使用此函数创建了新的线程，此线程就是subloop
// 在startLoop中，利用条件变量等待此函数创建好EventLoop
void EventLoopThread::threadFunc()
{
    EventLoop loop;

    if(Callback_)
    {
        Callback_(&loop);
    }

    {
        std::unique_lock<std::mutex> lock(mutex_);
        loop_ = &loop;
        cond_.notify_one();
    }

    loop.loop();
    std::unique_lock<std::mutex> lock(mutex_);
    loop_ = nullptr;
}

EventLoopThreadPool

EventLoopThreadPool 是一个线程池，封装了多个 EventLoopThread，并支持动态设置线程池的大小。当 MainLoop 接收到客户端连接后，EventLoopThreadPool 会根据负载均衡算法选择一个 SubLoop，由其负责该连接后续的所有事件处理。

EventLoopThreadPool.h

class EventLoopThreadPool : noncopyable
{
public:
    using ThreadInitCallback = std::function<void(EventLoop*)>;

    EventLoopThreadPool(EventLoop* baseloop, const std::string& name);
    ~EventLoopThreadPool();

    void setThreadNum(int numThreads) { numThreads_ = numThreads; }

    void start(const ThreadInitCallback& cb = ThreadInitCallback());

    // 如果工作在多线程，baseLoop_默认以轮询方式分配channel给subloop
    EventLoop* getNextLoop();

    std::vector<EventLoop*> getAllLoops();

    bool started() const { return started_; }
private:
    EventLoop* baseloop_; // mainloop
    std::string name_;
    bool started_;
    int numThreads_;
    int next_;
    std::vector<std::unique_ptr<EventLoopThread>> threads_;
    std::vector<EventLoop*> loops_;
};

解析：

成员变量：

baseloop_：MainLoop
threads_：存储了numThreads个的EventLoopThread
loops_：存储了numThreads个的loops_

成员方法：

setThreadNum()：设置线程的数量
start()：创建多个线程，并在线程内启动事件循环
getNextLoop：轮询算法获取EventLoop，成员变量next_表示下一个该取得EventLoop

EventLoopThreadPool.cc

EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseloop, std::string const &name) :
    baseloop_(baseloop),
    name_(name),
    started_(false),
    numThreads_(0),
    next_(0)
{}

EventLoopThreadPool::~EventLoopThreadPool() {}

void EventLoopThreadPool::start(ThreadInitCallback const &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));
        loops_.push_back(t->startLoop()); 
    }
    if(numThreads_ == 0 && cb)
    {
        cb(baseloop_);
    }
}

EventLoop *EventLoopThreadPool::getNextLoop()
{
    EventLoop* loop = baseloop_;

    if(!loops_.empty())
    {
        loop = loops_[next_++];
        if(next_ >= loops_.size())  next_ = 0;
    }

    return loop;
}

std::vector<EventLoop *> EventLoopThreadPool::getAllLoops()
{
    if(!loops_.empty())
    {
        return std::vector<EventLoop*>(loops_);
    }
    else
    {
        return loops_;
    }
}

解析：

void start(ThreadInitCallback const &cb)：方法的作用是创建多个线程，并在每个线程中启动事件循环，同时将线程及其对应的事件循环实例存储起来。

原文地址：https://blog.csdn.net/2404_87273268/article/details/144433541

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：Rust学习路线图
下一篇：美业多门店管理系统的【客户管理】有什么作用？必须具备哪些功能？

.NET(C#) 如何配置用户首选项及保存用户设置
.NET(C#) 如何配置用户首选项及保存用户设置
阅读更多2024-12-14
【最新】北大数字普惠金融指数数据集-省市县（2011-2023年）
郭峰,王靖一,王芳,孔涛,张勋,程志云.测度中国数字普惠金融发展:指数编制与空间特征[J].经济学(季刊),2020,19(04):1401-1418.时间跨度：省级和城市级指数时间跨度为2011-2
阅读更多2024-12-14
GESP202412 四级【Recamán】题解（AC）
a11ak−1−kkakak−1−kak−1k小杨想知道 Recamán 数列的前n项从小到大排序后的结果。手动计算非常困难，小杨希望你能帮他解决这个问题。
阅读更多2024-12-14
IDEA遇到EasyConnect中的网络资源无法访问的问题
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。原文链接：https://blog.csdn.net/wanshanyu_/article/de
阅读更多2024-12-14
双目摄像头标定方法
此时已经完成标定，左下角为反投影误差，右边为外参可视化。将双目左右目拍的图像上传（左右目最好不少于20张）此时回到主页面，即可看到成功导出。把这些误差大的删除即可。
阅读更多2024-12-14
Servlet、omcat服务器架构与工作原理
Servlet是运行在服务器端的Java程序，它的主要职责之一是接收并处理来自客户端（如浏览器）的HTTP请求。当客户端发送一个请求到服务器时，Servlet可以解析请求中的信息，例如请求的URL路径
阅读更多2024-12-14
Vue生命周期钩子函数：深入解析与实践
作为高级Vue前端开发人员，对Vue组件的生命周期钩子函数有着深刻的理解是至关重要的。生命周期钩子函数是指在Vue组件的创建、更新、销毁等过程中，Vue自动调用的一系列方法。通过这些钩子函数，我们可以
阅读更多2024-12-14
安卓开发--使用android studio发布APP
app发布
阅读更多2024-12-14
数据结构与算法学习笔记----拓扑排序
@ author: 明月清了个风。
阅读更多2024-12-14
python 将数据保存到现有的Excel文件的新工作表
out_file = ‘query.xlsx’df1 = pd.DataFrame(out_data)若直接写入：df1.to_excel(out_file, index=False, sheet_n
阅读更多2024-12-14