muduo

基于C++11的muduo网络库

作者：shenmingik

邮箱：2107810343@qq.com

时间：2021/1/26 22:17

开发环境：Ubuntu VS Code

编译器：g++

编程语言：C++

源码链接： 微云链接

写在前面

项目编译问题

项目编译时基于cmake的，在源码链接中有CMakeLists.txt文件，下载完之后直接点击那个编译即可。

库安装的问题

在源码链接的/lib目录中有一个autobuild.sh文件，用chmod +x autobuild.sh命令给它加上执行权限之后，执行就可。

注：博主的是Ubuntu系统，其他系统可以进去把地址给改一下

项目测试代码

这里简单的写了一个回显服务器用于测试，在源码链接的/example目录下，大家可以下载自己测试一下。

关于压力测试

由于我这里基本就是重写了一下原来的muduo + 懒（不是主要原因）

所以

压力测试可以直接参考陈硕大神做的：

muduo 与 libevent2 吞吐量对比 muduo 与 boost asio 吞吐量对比

项目概述

这个项目呢，是将陈硕大神的muduo网络库源码中核心代码部分重新写了一遍，将原来依赖boost库的地方都替换成了C++ 11语法。算是博主对muduo网络库达成更好理解的一个产品吧。

muduo网络库的reactor模型

在muduo网络库中，采用的是reactor模型，那么，什么是reactor模型呢？

Reactor： 即非阻塞同步网络模型，可以理解为，向内核去注册一个感兴趣的事件，事件来了去通知你，你去处理 Proactor： 即异步网络模型，可以理解为，向内核去注册一个感兴趣的事件及其处理handler，事件来了，内核去处理，完成之后告诉你

muduo的设计

reactor模型在实际设计中大致是有以下几个部分：

• Event：事件
• Reactor： 反应堆
• Demultiplex：多路事件分发器
• EventHandler：事件处理器

在muduo中，其调用关系大致如下

• 将事件及其处理方法注册到reactor，reactor中存储了连接套接字connfd以及其感兴趣的事件event
• reactor向其所对应的demultiplex去注册相应的connfd+事件，启动反应堆
• 当demultiplex检测到connfd上有事件发生，就会返回相应事件
• reactor根据事件去调用eventhandler处理程序

而上述的，是在一个reactor反应堆中所执行的大致流程，其在muduo代码中包含关系如下（椭圆圈起来的是类）：

可以看到，EventLoop其实就是我们的reactor，其执行在一个Thread上，实现了one loop per thread的设计。 每个EventLoop中，我们可以看到有一个Poller和很多的Channel，Poller在上图调用关系中，其实就是demultiplex（多路事件分发器）,而Channel对应的就是event（事件）

现在，我们大致明白了muduo每个reactor的设计，但是作为一个支持高并发的网络库，单线程 往往不是一个好的设计。

muduo采用了和Nginx相似的操作，有一个main reactor通过accept组件负责处理新的客户端连接，并将它们分派给各个sub reactor，每个sub reactor则是负责一个连接的读写等工作。

muduo各个类

明白了muduo的细节之后，我们对muduo的剖析就更为容易。

辅助类

这个类别的类与网络实现没有太大关系，只是用来辅助网络库的实现了

NonCopyable

这个类将拷贝和赋值构造函数给delete掉，提供了一个不可拷贝的基类

    NonCopyable(const NonCopyable &) = delete;
    NonCopyable &operator=(const NonCopyable &) = delete;

TimeStamp

这个类用于给网络库提供系统时间，我这里用的是time（nullptr）函数

    TimeStamp();
    explicit TimeStamp(int64_t times);
    //获取当前时间
    static TimeStamp now();
    //转换为字符串
    string to_string();

Logger

这个是日志类,采用的是饿汉式的单例模式，用于打印网络库运行过程中的日志信息，主要分为四个级别

• INFO: 正常的日志输出
• ERROR: 有错误的日志输出，但是程序还可以运行
• FATAL: 有错误的日志输出，程序不可运行，直接exit
• DEBUG: 用于调试得到错误信息

同时也往外提供了四个宏函数用于打印信息：LOG_INFO、 LOG_ERROR、LOG_FATAL、LOG_DEBUG，由于四个函数相似度较大，我这里就放出一个函数LOG_INFO

#define LOG_INFO(logmsgFormat, ...)                       \
    do                                                    \
    {                                                     \
        Logger &logger = Logger::instance();              \
        logger.set_log_level(EnLogLevel::INFO);           \
        char buf[BUFFER_SIZE] = {0};                      \
        snprintf(buf, 1024, logmsgFormat, ##__VA_ARGS__); \
        logger.log(buf);                                  \
    } while (0)

我们可以看到，这里是对输入数据进行了处理然后调的logger的log函数进行打印。

    //获取唯一实例对象
    static Logger &instance();

    //设置日志级别
    void set_log_level(EnLogLevel level);

    //写日志
    void log(string msg);

Buffer

这个是muduo网络库中底层的数据缓冲类型，模仿java中netty的设计，其有一个prepend、read、write三个标志，划分了缓冲区的数据。 其中perpend-read之间是一个头部的标志位，read-write是可读数据，write-末尾是可写数据。

应用将数据写入到网络库的Buffer缓冲区，然后Buffer缓冲区再写到TCP的缓冲区，最后再发送。

数据：
    vector<char> buffer_;
    size_t read_index_;
    size_t write_index_;
方法：
    //返回可读的长度
    size_t readable_bytes() ;
    //返回可写的长度
    size_t wirteable_bytes();
    //返回头长度
    size_t prependable_bytes();
    //返回缓冲区中可读数据的起始地址
    const char *peek();
    //缓冲区readindex 偏移
    void retrieve(size_t len);
    //缓冲区复位
    void retrieve_all();
    //读取所有数据
    string retrieve_all_asString();
    //读取len长度数据
    string retrieve_as_string(size_t len);
    //保证缓冲区有这么长的可写空间
    void ensure_writeable_bytes(size_t len);
    //返回可写数据地址
    char *begin_write();
    //忘缓冲区中添加数据
    void append(const char *data, size_t len);
    //从fd中读取数据
    ssize_t readfd(int fd, int *save_errno);
    //通过fd发送数据
    ssize_t writefd(int fd, int *save_errno);
private:
    //返回缓冲区起始地址
    char *begin();
    //扩容函数
    void makespace(size_t len);

Reactor中类

这个类别中主要讲解reactor中要实现的类

InetAddress

这个类封装了socket所要绑定的ip地址和端口号，比较简单

    string get_ip() const;
    string get_ip_port() const;
    uint16_t get_port() const;

    void set_sockaddr(const sockaddr_in &addr) { addr_ = addr; }
    const sockaddr_in *get_sockaddr() const { return &addr_; }

Channel

这个类中主要是封装了sockfd及其所感兴趣的事件，还有发生事件所要调用的回调函数。

数据：
    EventLoop *loop_;	//所属loop
    const int fd_;
    int events_;      //fd感兴趣事件
    int real_events_; //poller 具体发生的事件
    int index_;

    weak_ptr<void> tie_; //观察当前channel的存在状态
    bool tied_;          //判断tie_是否绑定过

    //发生事件所要调用的具体事件的回调操作
    ReadEventCallback read_callback_;
    EventCallback write_callback_;
    EventCallback close_callback_;
 方法：
     //fd得到poller通知以后，根据具体发生的事件，调用相应的回调
     //其实调用的handle_event_withGuard
    void handle_event(TimeStamp receive_time);
    //防止channel被remove掉，channel还在执行回调
    //一个tcpconnection新连接创建的时候，调用tie
    void tie(const shared_ptr<void> &);
    //得到socket套接字
    int get_fd();
    //得到感兴趣事件
    int get_events();
    //设置真正发生的事件,poller监听到事件然后设置real_event
    int set_revent(int event);
    //判断该fd是否设置过感兴趣事件
    bool is_noneEvent();
    //返回所属eventloop
    EventLoop *owner_loop();
    //在channel所属的eventloop中删除自己
    void remove();
public:
    //设置fd感兴趣事件
    void enable_reading();
    void dis_enable_reading();
    void enable_writing();
    void dis_enable_writing();
    void dis_enable_all();
public:
    //返回fd当前感兴趣事件状态
    bool is_none_event();
    bool is_writting();
    bool is_reading();
public:
    //设置发生不同事件的回调操作
	......
private:
    //与poller更新fd所感兴趣事件
    void update();
    //根据发生的具体事件调用相应的回调操作
    void handle_event_withGuard(TimeStamp receive_time);

EpollPoller

这个封装了epoll，也就是底层的demultiplex（多路事件分发），里面包含了一个指向Channel的指针，以及自己在内核事件表中的fd

数据：
    int epollfd_;
    //key: fd  value:fd所属channel
    ChannelMap channels_;		//unordered_map<int, Channel *>;
方法：
	//调用epoll_wait，并将发生事件的channel填写到形参active_channel中
	TimeStamp poll(int timeout, ChannelList *active_channels) override;
	//往channel_map中添加channel
    void update_channel(Channel *channel);
    //channel_map中删除channel
    void remove_channel(Channel *channel);
private:
    //填写活跃的链接
    void fill_active_channels(int events_num, ChannelList *active_channels) const;
    //更新channel，调用epoll_ctl
    void update(int operation, Channel *channel);

EventLoop

这个是事件循环类，主要包含两个组件 ----- Poller以及Channel。

数据：
    atomic_bool looping_;
    atomic_bool quit_; //标志退出loop循环

    const pid_t threadId_;       //记录当前loop所在线程id
    TimeStamp poll_return_time_; //poller返回的发生事件的时间点
    unique_ptr<Poller> poller_;

    int wakeup_fd;                       //当main loop获取一个新用户的channel，通过轮询算法，选择一个subloopp，通过该成员唤醒subloop，处理channel
    unique_ptr<Channel> wakeup_channel_; //包装wakefd

    ChannelList active_channels; //eventloop 所管理的所有channel

    atomic_bool calling_pending_functors_; //标识当前loop是否有需要执行的回调操作
    vector<Functor> pending_Functors_;     //loop所执行的所有回调操作

    mutex functor_mutex_; //保护pending_functors

可以看到，其有一个指向Poller的指针，以及一个存储Channel的容器，ChannelList。

值得注意的是，这里还有一个wakeup_fd，这是干什么的呢？

唤醒 eventloop用的！

试想一下，如果一个eventloop一直处于epoll_wait的阻塞状态，那么我main reactor怎么去给他分配新的连接？我不得叫醒它？

在libevent中也有相似的组件，不过用的是sockpair，相当于在本地创建socket进行通信，而muduo中用的则是更加高效的eventfd（事件驱动，更快，8字节缓冲区，更省）。上层模块通过往eventfd中去写，触发写事件，内核就自然而然的将相应的eventloop唤醒了。

方法：
    //开启事件循环
    void loop();
    //退出事件循环
    void quit();
    TimeStamp get_poll_returnTime() const { return poll_return_time_; }
    //在当前loop中执行cb
    void run_in_loop(Functor cb);
    //把cb放入队列中,唤醒loop所在线程执行cb（pending_functor）
    void queue_in_loop(Functor cb);
    //唤醒loop所在线程
    void wakeup();
    //poller的方法
    void update_channel(Channel *channel);
    void remove_channel(Channel *channel);
    bool has_channel(Channel *channel);
    //判断eventloop对象是否在自己的线程中
    bool is_in_loopThread() const;
private:
    void handle_read();         //wake up
    void do_pending_functors(); //执行回调

现在我们大概对已经介绍的类有了一点眉目了，其实也就是对epoll的一个封装，原来的EpollLoop在epoll_create，注册各个channel之后，就处于epoll_wait处于阻塞状态。如果这个时候，之前的channel没有事件发生，而上层又想唤醒当前的EventLoop去执行新的连接，就调用wakeup，唤醒当前的EventLoop。

Thread

这个类，在原来的muduo中使用linux系统调用pthread_create那样写的较为繁琐，这里直接使用了C++ 11的thread类。相对来说，也比较简单了。

数据：
    bool started_;
    bool joined_;
    shared_ptr<thread> thread_;

    pid_t tid_;
    ThreadFunc function_;
    string name_;
方法：
    void start();
    void join();
    bool is_started() const { return started_; }
    pid_t get_tid() const { return tid_; }
    const string &get_name() const { return name_; }
    static int get_thread_nums() { return thread_nums_; }
private:
    void set_default_name();

EventLoopThread

都说muduo网络库的核心是one loop per thread，即一个线程一个eventloop。其实现的秘密就蕴藏在EventLoopThread类中。

数据：
    EventLoop *loop_;
    bool exiting_;
    Thread thread_;
    mutex thread_mutex_;
    condition_variable condition_;
    ThreadInitCallback callback_function_;
方法：
    EventLoopThread(const ThreadInitCallbac