基于C++11重写muduo库

Posted 2021-06-01 wtxuebc

邮箱：[e-mail]

源码：基于C++11重写muduo库

如有转载，请标明出处，谢谢！

文章目录

• 预备知识
• • muduo库简介
  • muduo库的优点
  • • • muduo和libevent
      • one loop per process
• 言归正传
• • muduo的模型设计
  • • • Reactor模型
      • muduo库的Reactor模型
      • one (evnet)loop per thread
• C++11重写muduo核心类代码
• • muduo库的核心类代码
  • • 辅助类
    • • Noncopyable
      • Timestamp
      • Logger
      • socket
      • InetAddress
      • Buffer
      • CurrentThread
    • TcpServer以及其相关类
    • • TcpServer
      • Acceptor
      • TcpConnection
    • EventLoop以及其相关类
    • • Channel
      • Poller
      • EpollPoller
      • EventLoop
      • EventLoopThread
      • EventLoopThreadPool
• 总结

预备知识

muduo库简介

​ muduo库是一个高质量的基于Reactor模型的开源网络库，采用one loop per thread + thread pool 的架构来实现的,只支持linux系统；

​ 这里附上作者陈硕大牛的介绍发布一个基于 Reactor 模式的 C++ 网络库

muduo库的优点

既然是one loop per thread就从两方面比，one loop 和 per thread

muduo和libevent

笔者libevent只是简单用过，没有去深入的刨析源代码，以下如有不对，请不吝指出！

libevent使用c写的一个网络库也是基于Reactor模型，它没有用多个loop的这个思想，整体就是一个大的Reactor，自然也没有分发链接的过程，libevent主线程和io线程中间使用socketpair通信的；那换一句话说，就是muduo的one loop、一个线程一个loop有啥优缺点

思考一下可以大体知道，当sockfd上有事件发生（可能是新用户链接也可能是读写事件），所有的线程都会竞争这个事件的执行权，但是最后只有一个线程执行这个操作，显得其他线程无事可做倒有点浪费资源；

而muduo库有baseLoop去执行新用户的链接，其他ioLoop只负责固定连接的读写事件，可以并发执行多个连接的读写事件，充分利用CPU

陈硕大牛是这样说的：

multi-reactors可以处理突发IO（利用多线程处理多个连接上的IO）。

总结下来就是：计算量大，IO带宽小，去用单个eventloop。如果IO带宽大，计算量小，则用多个eventloop。对延时敏感也是用多个eventloop。

由于本文重点刨析muduo源码，这些知识等笔者再去深入研究后再隔篇重讲！

one loop per process

one loop per process就是一个进程一个事件循环，这个问题归根结底就是讨论多进程和多线程的区别

虽然线程比进程消耗小，但是不一定比进程快

总结下来就是：进程间的切换消耗是大于线程的，但是一个进程阻塞了不影响另一个线程，同一个进程的一个线程阻塞了则会影响其他线程；

举个例子：两段公路A和B，车子在A上面跑虽然耗油，但是没红绿灯，跑得快；

​ 而车子在B上面跑虽然省油，但是红绿灯多，跑的相对而言较慢；

两者没有最好的，只有用在不同场景能体现其对应的价值！

言归正传

muduo的模型设计

Reactor模型

首先得明白reactor和proactor模型的区别

reactor：用于同步IO，同步IO就是说，当内核缓冲区有数据时，IO函数被唤醒，IO函数一执行，开始从内 ​ 核缓冲区向用户缓冲区拷贝数据，所耗费的时间都记在程序用时；

proactor：用于异步IO，异步IO就是说用户把sockfd、buf、和通知方式（不止信号还有回调）通过IO接口
​ 传给内核，当内核缓冲区有数据时，IO函数被唤醒，IO函数一执行内核缓冲区就有数据；内核会
​ 把数据拷贝到用户提供的buf中，并按照用户提供的通知方式应用程序，这个拷贝的过程，全程都
​ 是内核完成的，不计入应用程序耗时；

Reactor是由**Event(事件)、Reactor(反应堆)、Demultiplex(事件分发器)和Evanthandler(事件处理器)**四部分构成的

Event：注册事件和其对应的处理方法给Reactor，Reactor里边有sockfd以及其对应的event

Reactor：向其Demultiplex添加 ( 此处以add为例，其他方法也一样 )，启动反应堆

Demultiplex：开启事件循环，将发生事件的event返回给Reactor

EventHandler：Reactor调用event对应的事件处理器EventHandler执行相应的回调函数

muduo库的Reactor模型

• channel封装了Event
• EventLoop就是一个Reactor
• Poller就是一个Demultipex
• channel里边有对应的回调函数
• EvnetLoop在一个单独的进程里边执行(one (event)loop per thread)
• 一个EventLoop里边有一个Poller和多个Channel，其中Poller和channel不能相互调用，必须通过EventLoop来进行相关数据的传输
• channel需要调用EventLoop的方法，对channelMap里边的channel进行增删改查
• Poller监听sockfd，并把返回的有事件发生的fd对应的channel添加到activechannel里边
• EventLoop通知channel执行对应的回调

one (evnet)loop per thread

​ 在muduo库里边有两种线程：一种里边的事件循环专门处理新用户连接，一种里边的事件循环专门处理对应连接的所有读写事件

mainLoop( 也就是baseLoop): 负责处理新用户的连接

• Acceptor主要封装了listenfd( 对应的新用户连接事件 )的相关操作，将listenfd封装成了acceptChannel，并注册到其Poller中，Poller监听到事件发生后，mainLoop通知acceptChannel执行相关回调（此回调是Acceptor传给的）

• 执行Acceptor::handleread时，先进行一个accpte操作，再执行一个回调，newConnectioncallback是TcpServer传给的，轮询选择一个subloop

• 将connfd封装成channel分发给subloop，此loop负责此connfd上的所有操作；

  ioLoop：负责处理连接上的各种操作；包括建立连接、断开连接以及发送数据、接收数据

​ muduo库底层实现了一个buffer，负责数据的存储，用于接收和发送

C++11重写muduo核心类代码

muduo库的核心类代码

辅助类

Noncopyable

此类主要做的就是通过下面两行代码，让其派生类不能进行拷贝构造和赋值，让代码更易读

      noncopyable(const noncopyable&) = delete;
      //让基类不能进行拷贝构造，从而使派生类也不能进行拷贝构造
      noncopyable& operator=(const noncopyable&) = delete;
      //让基类不能进行赋值运算，从而使派生类也不能进行赋值运算

Timestamp

此类主要为muduo库的一些log提供当前时间

     Timestamp();
     explicit Timestamp(int64_t microSecondsSinceEpoch);
     static Timestamp now();
     //获取当前的时间;直接return Timestamp(time(nullptr))
     std::string toString() const;

Logger

此类为一些操作提供相应日志信息，方便进行错误的追踪，采用的是饿汉模式(线程安全)

    INFO,  //普通信息
    ERROR, //错误信息
    FATAL, //core信息
    DEBUG, //调试信息

INFO：打印一些重要的流程信息

ERROR：不影响程序继续执行的错误信息

FATAL：影响程序继续执行的致命信息

DEBUG：调试信息

一共有四种级别的信息，每个函数都有其对应的宏函数LOG_INFO、LOG_ERROR、LOG_FATAL、LOG_DEBUG

#define [对应的宏](logmsgFormat, ...) \\
    do \\
    { \\
        Logger &logger = Logger::instance();  \\
        logger.setLogLevel([信息类型]);   \\
        char buf[1024] = {0}; \\
        snprintf(buf,1024,logmsgFormat,##__VA_ARGS__); \\
        logger.log(buf); \\
    } while(0) 

socket

该类主要封装了sockfd对应的操作

    int fd() { return sockfd_; }
    void bindAddress(const InetAddress &localaddr);
    void listen();
    int accept(InetAddress *peeraddr);

InetAddress

该类主要封装了socket的地址信息，可以得到ip地址和端口号

    explicit InetAddress(uint16_t port = 0,std::string ip = "127.0.0.1");
    explicit InetAddress(const sockaddr_in &addr)
        :addr_(addr)
        {}
    std::string toIp() const;   //得到ip
    std::string toIpPort() const;    //得到ip加端口号
    uint16_t toPort() const;   //得到端口号
    
    const sockaddr_in* getSockAddr() const{return &addr_;}
    void setSockaddr(const sockaddr_in &addr) { addr_ = addr;}

Buffer

muduo库底层的发送、接收的消息都是存放在Buffer缓冲区的

    static const size_t kCheapPrepend = 8;   //用来初始化首部信息
    static const size_t kInitialSize = 1024;   //用来初始化数据区

//构造函数
explicit Buffer(size_t initialSize = kInitialSize)     //initialSize是数据区大小·
             :buffer_(kCheapPrepend + kInitialSize)    //缓冲区的大小等于首部加数据区的大小
             ,readerIndex_(kCheapPrepend)              //可读区域的首地址下标
             ,writerIndex_(kCheapPrepend)              //可写区域的首地址下标
             {}

各种关系大体都是围绕这几个变量计算的

    //可读的大小
    size_t readableBytes() const 
    {
        return writerIndex_ - readerIndex_;
    }

    //可写的大小
    size_t writableBytes() const
    {
        return buffer_.size() - writerIndex_; 
    }

CurrentThread

该类通过系统调用获取当前线程的tid

    extern __thread int t_cachedTid;

    void cacheTid()
    {
       if (t_cachedTid == 0)
       {  
          //通过系统调用获取了当前线程的tid值
          t_cachedTid = static_cast<pid_t>(syscall(SYS_gettid)); 
       }
    }

    inline int tid()
    {
        //如果线程tid等于0就调用这个（通过系统调用获取线程tid的函数）来获取线程的tid
        //如果线程tid不等于0就返回它自己的；
        if (__builtin_expect(t_cachedTid == 0,0))
        {
           cacheTid();
        }
        return t_cachedTid;
    }

TcpServer以及其相关类

TcpServer

    void newConnection(int sockfd, const InetAddress &peerAddr);  //增加新连接
    void removeConnection(const TcpConnectionPtr &conn);  //删除链接
    void removeConnectionInloop(const TcpConnectionPtr &conn); //在一个loop中删除该链接
 
    using ConnectionMap = std::unordered_map<std::string, TcpConnectionPtr>; //连接的名字和对应的连接


    EventLoop *loop_;     //默认的loop，即baseloop
    const std::string ipPort_;   //ip和端口号
    const std::string name_;    //名字

    std::unique_ptr<Acceptor> acceptor_;  //专门负责处理新用户的链接
    std::unique_ptr<EventLoopThreadPool> threadPool_;   //one loop per thread;

    ConnectionCallback connectionCallback_;  //有新链接的回调
    MessageCallback messageCallback_;      //有读写事件的回调
    WriteCompleteCallback writeCompleteCallback_; //消息发送完成以后的回调

    ThreadInitCallback threadInitCallback_; //loop线程初始化的回调

    std::atomic_int started_;

    int nextConnId_;
    ConnectionMap connections_; //保存所有的链接

    TcpServer(EventLoop *loop,     //默认的loop，即Baseloop  
              const InetAddress &listenAddr,    //地址信息
              const std::string &nameArg,    //名字
              Option option = kNoReusePort);   //端口是否重用

构造函数，主要传的参数就是前三个

    void setThreadInitcallback(const ThreadInitCallback &cb){ threadInitCallback_ = cb; }
//线程初始化回调函数
    void setConnectionCallback(const ConnectionCallback &cb){ connectionCallback_ = cb; }  //链接事件回调函数
    void setMessageCallback(const MessageCallback &cb) { messageCallback_ = cb; }
//读写事件回调函数
    void setWriteCompleteCallback(const WriteCompleteCallback &cb)          {writeCompleteCallback_ = cb; }

其中新用户链接的回调会传给Accptor

​ 读写事件的回调会传给TcpConnection

TcpServer主要的做的事就是：

• setThreadNum

void TcpServer::setThreadNum(int numThreads) 
  //设置底层EventLoop的数量，就是ioLoop的数量，若数量为零，那connfd的读写事件也由baseLoop执行
{
   threadPool_->setThreadNum(numThreads);
}

• start

void TcpServer::start() //开启服务监听
{
   if(started_++ ==0)  //防止一个TcpServer对象被start多次
   {
       threadPool_->start(threadInitCallback_);    //启动底层线程池
       loop_->runInloop(std::bind(&Acceptor::listen,acceptor_.get()));
       //启动acceptor所在的loop，并且acceptor开始监听新用户的链接
   }
}

还有就是它负责实现Acceptor和TcpConnection所要调用回调

Acceptor

此类主要处理新用户的链接，并且链接得到connfd，这个connfd里边的读写事件由TcpConnection执行

还是这个图，Acceptor是在baseLoop里边执行的

void handleRead();
EventLoop *loop_;   //Accptor用的就是用户定义的那个baseLoop，也称mainLoop
Socket acceptSocket_;  //它负责监听的就是acceptSocket_上的事件
Channel acceptChannel_;   //将acceptSocket_封装成acceptChannel_
NewConnectionCallback newConnectionCallback_;   //当loop_里的poller发现acceptSocket_上有事件发生时，向loop_返回acceptChannel_，Acceptor执行此回调
bool listenning_;

acceptor的核心就是handleRead

void Acceptor::handleRead()
{
   InetAddress peerAddr;
   int connfd = acceptSocket_.accept(&peerAddr);
   if(connfd >= 0)
   {
       if(newConnectionCallback_)
       {
           newConnectionCallback_(connfd,peerAddr);
       }
       else
       {
           ::close(connfd);
       }
   }
   else{
       LOG_ERROR("%s:%s:%d accept err:%d \\n",__FILE__,__FUNCTION__,__LINE__, errno);
       if(errno == EMFILE)
       {
          LOG_ERROR("%s:%s:%d reached limit! \\n",__FILE__,__FUNCTION__,__LINE__, errno);
       }
   }

}

当有新用户链接时执行这个回调，先accpet得到connfd，再执行newConnectioncallback（通过轮询的方式将connfd分发给ioLoop），当然这个回调都是传的TcpServer的方法，是在TcpServer里边实现的；

TcpConnection

此类主要是负责connfd的各种操作，因此回调比较多，都是TcpServer传给的；

    ConnectionCallback connectionCallback_;  //有新链接的回调
    MessageCallback messageCallback_;      //有读写事件的回调
    WriteCompleteCallback writeCompleteCallback_; //消息发送完成以后的回调
    HighWaterMarkCallback highWaterMarkCallback_; //高水位线
    CloseCallback closeCallback_;

    EventLoop *loop_;  //这里的loop绝对不是baseloop，因为TcpConnection都是在subloop里边管理的
    const std::string name_;  //Tcpconnection的名字
    std::atomic_int state_;   //链接的状态
    
    enum StateE {kDisconnected //已经断开的
                ,kConnecting  //正在链接
                ,kConnected  //已经连接上了
                ,kDisconnecting};  //正在断开 
    bool reading_;

由于Buffer是在发送、接收数据时用到的，因此Buffer是在TcpConnection里面用到的

EventLoop以及其相关类

Channel

此类负责打包sockfd（acceptfd和connfd）,不同fd对应的loop提醒channel执行的回调也是不同的

acceptfd的回调是传的Acceptor的，对应的connfd的回调是传的TcpConnection的

它的主要成员变量如下

   EventLoop *loop_;  //事件循环（所属的事件循环）
   const int fd_;   //fd，就是poller监听的对象
   int events_;  //注册fd感兴趣的事件
   int revents_;  //发生的事件
   int index_;  //表示的就是channel的状态（未添加、已添加和已删除）
   //新创建的channel默认值为-1，通过updatechannel和removechannel的一个操作改变其对应的值

   void update();
   void remove();
   void handleEventWithGuard(Timestamp receiveTime);  //handle(处理) Event(事件) WithGuard(受保护的)

还有对Poller里边map表里边的channel_的管理是通过EventLoop对Poller执行的

因此这几个方法底层直接调用的EventLoop的对应方法

channel就主要是封装了fd和其对应的回调

   //因为Channel通道里边能够获知fd最终发生的具体的事件revent，所以它负责调用具体事件的回调操作；
   ReadEventCallback readCallback_;
   EventCallback writeCallback_;
   EventCallback closeCallback_;
   EventCallback errorCallback_;

Poller

此类是一个抽象接口类，向下提供不同的IO复用函数的接口

    //给所有的IO复用保留统一的接口,需要它们去重写的;
    //当前激活的channel，运行的channel，需要Poller去照顾的channel都在ChannelList里边;
    virtual Timestamp poll(int timeoutMs,ChannelList *activeChannels) = 0;
    virtual void updateChannel(Channel *channel) = 0;
    virtual void removeChannel(Channel *channel) = 0;

    using ChannelMap = std::unordered_map<int,Channel*>;  
    //poller监听的channel是EventLoop（EventLoop里边有ChannelList）里边保存的channel

EpollPoller

此类是用EPollio复用实现的Poller，主要管理和监听channel，并将sockfd有事件发生的channel返回给EventLoop，由EventLoop通知channel执行相应的回调

    //还负责重写基类的方法
    static const int kInitEventListSize = 16;  //保存事件的vector初始化大小
    
    //填写活跃的channel在activeChannel(ChannelList)里边
    void fillActiveChannels(int numEvents, ChannelList *activeChannel) const;
    //更新channel通道
    void update(int operation, Channel *channel);

    using EventList = std::vector<epoll_event>;  //事件集合

    int epollfd_;
    EventList events_; 

EventLoop

此类相当于一个Reactor的角色，向Poller的map表里边注册fd事件对应的channel，并接受有事件发生的channel，并通知channel执行相应的回调

    void handleRead();  //wakeup  通过给wakeupfd上写入数据，唤醒ioLoop
    void doPendingFunctors();  //执行回调

    using ChannelList = std::vector<Channel*>;   
    std::atomic_bool looping_;  //原子操作，底层通过CAS实现的
    std::atomic_bool quit_;  //标志退出loop循环

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

    int wakeupFd_; //主要作用，当mainloop获取一个新用户的channel，通过轮训算法选择一个subloop，通过该成员唤醒subloop处理
    std::unique_ptr<Channel> wakeupChannel_;

    ChannelList activeChannels_;
    Channel *currentActiveChannel_;

    std::atomic_bool callingPendingFunctors_;  //标识当前loop是否需要执行的回调操作
    std::vector<Functor> pendingFunctors_;  //存储loop需要执行的所有的回调操作
    std::mutex mutex_; //互斥锁，用来保护上面vector容器的线程安全操作

至此muduo库重要的组件以及每个EventLoop实现的功能已经很清楚了，现在需要考虑的就是在多线程下，实现one loop per thread

EventLoopThread

此方法是让一个loop在单独创建的线程中执行

    EventLoop *loop_;   //对应的loop
    bool exiting_;      //线程的状态
    Thread thread_;     //对应的线程
    std::mutex mutex_;   
    std::condition_variable cond_; 
    ThreadInitCallback callback_;

   
EventLoop* EventLoopThread::startLoop()
{
   thread_.start();  //loop开启时，底层的线程开启了，回调函数threadFunc()就开启了

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

//此方法是在单独的线程中执行的
void EventLoopThread::threadFunc()
{
   EventLoop loop;  //创建一个独立的eventloop，和上面的线程一一对应

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

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

   loop.loop();  //loop会一直执行，到下一步说明loop已经结束循环了；
   
   std::unique_lock<std::mutex> lock(mutex_);
   loop_ = nullptr;
}

通过条件变量可以保证创建的线程和运行的loop一一对应

EventLoopThreadPool

还记得刚开始TcpServer设置了ThreadNum，在这个类中会根据设置的数量开启相应的线程，每个线程都和一个ioLoop一一对应，并且阻塞在recv等待被唤醒；

    EventLoop *baseLoop_;
    std::string name_;
    bool started_;
    int numThreads_;
    int next_;     //做下一个loop的下表用的；
    std::vector<std::unique_ptr<EventLoopThread>> threads_;  //包含了Loop线程
    std::vector<EventLoop*> loops_;  //包含了所有loop的指针; (还记得调用EventLoopThread里边的startLoop()就能得到loop指针)

    //TcpServer的两个方法底层就是调用者俩个方法就是调用的
    void setThreadNum(int numThreads) { numThreads_ = numThreads;}     
    void start(const ThreadInitCallback &cb = ThreadInitCallback())
    {
    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());  //执行startLoop函数，返回loop指针，（没执行threadFunc灰调函数的话，线程会阻塞）
                           //底层创建线程，绑定一个新的EventLoop,并返回该loop的地址
    }
    if(numThreads_ == 0 && cb)
    {
        cb(baseLoop_);       //cb(ThreadInitCallback)（用户提前设置的回调）
    }
    }

在Acceptor的HandRead回调函数中，得到connfd，然后通过轮询的方法得到下一个loop，底层也是调用的Pool里边的GetNextLoop()方法；

//如果工作在多线程中，baseLoop_默认以轮训的方式纷飞channel给subloop
EventLoop* EventLoopThreadPool::GetNextLoop()
{
    EventLoop *loop = baseLoop_;
    if(!loops_.empty())
    {
       loop = loops_[next_];
       next_++;
       if(next_ >= loops_.size())
       {
        next_ = 0;
       }
    }
    return loop;
}

总结

核心代码到这里就基本拉下序幕了，我们可以再总结一下

• muduo库采用one loop per thread + thread pool的模型；

• 总体是基于Reactor模型的（EventLoop就是一个Reactor、Poller就是一个事件分发器）;

  sockfd以及其对应的回调操作被打包成channel，由EventLoop将其添加到Poller的map表里边，Poller将发生事件的channel返回在EventLoop的List表里边，EventLoop通知channel执行相应的回调操作；

• muduo库的TcpServer负责传入用户写的回调操作；

  此时fd分为两种p（acceptfd和connfd），分别和其对应的事件打包成channel，acceptfd感兴趣的事件是新用户的链接事件，因此acceptChannel被添加到baseLoop的Poller里边；

  当有新用户链接时，执行acceptfd对应的回调操作，并得到connfd，此时将connfd和其感兴趣的事件添加到ioLoop的Poller里边，当对应的事件发生时，调用相关的回调操作！

参考文献：施磊《腾讯课堂》