手写一个moduo网络库

2025-04-01

字数统计: 4.7k字 | 阅读时长≈ 19分

学习手写moduo网络库

本项目是通过学习网络库moduo，学习优秀的代码设计，掌握基于事件驱动和回调的epoll，线程池面相对象编程

用C++ 14手写一遍网络库，不依赖boost
学习来源：
施磊老师的教程，照着视频敲代码
https://www.bilibili.com/video/BV1UE4m1R72y/?spm_id_from=333.1387.favlist.content.click&vd_source=502a85946570f43a2e447526db3235ab

Muduo库是基于Reactor模式实现的TCP网络编程库。该文章后续篇幅都是围绕Multi-reactor模型进行展开。Multi-Reactor模型如下所示

项目进度：

核心类组件
TcpServer服务端
TcpClient编写客户端类（进行中…）
支持定时事件TimerQueue 链表/队列 (后续学习)
DNS、HTTP、RPC（后续学习）
丰富的使用示例examples目录（后续学习）

基础知识：

推荐阅读：《Linux高性能服务器编程》

陈硕官方: https://github.com/chenshuo/muduo/

典型的一次IO的两个阶段是什么？数据准备和数据读写

数据准备：根据系统IO操作的就绪状态

阻塞

非阻塞

数据读写：根据应用程序和内核的交互方式

同步

异步

陈硕大神原话：在处理 IO 的时候，阻塞和非阻塞都是同步 IO。只有使用了特殊的 API 才是异步

IO。

Unix/linux 五中IO模型

阻塞blocking, 非阻塞 -non-blocking， IO复用，信号驱动，异步

好的模型：non-blocking + IO-multiplexing

select：

将已经连接的socket都放到一个文件描述符集合中，然后调用select函数将文件描述符拷贝到内核里，让内核检查网络事件产生（就是遍历一遍），当检查到有网络事件后，将次cokset标志为可读或可写，接着把整个文件描述符拷贝到用户态里，然后用户态还需要遍历，可读可写的socket，然后对其处理（2次遍历，两次拷贝）

自己理解的大白话（面试还是专业些）：

我们可以开设一个代收网点，让快递员全部送到代收点。这个网店管理员叫select。这样我们就可以在家休息了，麻烦的事交给select就好了。当有快递的时候，select负责给我们打电话，期间在家休息睡觉就好了。

但select 代收员比较懒，她记不住快递员的单号，还有快递货物的数量。她只会告诉你快递到了，但是是谁到的，你需要挨个快递员问一遍。

while true {
    select(流[]); //阻塞

  //有消息抵达
    for i in 流[] {
        if i has 数据 {
            读 或者 其他处理
        }
    }
}

poll和select类似只是不用BitMap存储文件描述符，取而代之的是动态数组
select模式的缺点：

1、单个进程能够监视的文件描述符的数量存在最大限制，通常是1024，当然可以更改数量，但由于

select采用轮询的方式扫描文件描述符，文件描述符数量越多，性能越差；(在linux内核头文件中，有

这样的定义：#define __FD_SETSIZE 1024

2、内核 / 用户空间内存拷贝问题，select需要复制大量的句柄数据结构，产生巨大的开销

3、select返回的是含有整个句柄的数组，应用程序需要遍历整个数组才能发现哪些句柄发生了事件

4、select的触发方式是水平触发，应用程序如果没有完成对一个已经就绪的文件描述符进行IO操作，

那么之后每次select调用还是会将这些文件描述符通知进程

相比select模型，poll使用链表保存文件描述符，因此没有了监视文件数量的限制，但其他三个缺点依

然存在。

以select模型为例，假设我们的服务器需要支持100万的并发连接，则在__FD_SETSIZE 为1024的情况

下，则我们至少需要开辟1k个进程才能实现100万的并发连接。除了进程间上下文切换的时间消耗外，

从内核/用户空间大量的句柄结构内存拷贝、数组轮询等，是系统难以承受的。因此，基于select模型的

服务器程序，要达到100万级别的并发访问，是一个很难完成的任务。
epoll:

与select，poll一样，对I/O多路复用的技术
只关心“活跃”的链接，无需遍历全部描述符集合
能够处理大量的链接请求(系统可以打开的文件数目)
使用红黑树跟踪进程所有待检测的文件描述符
使用时间驱动机制，内核里维护了一个链表记录事件，当我某个socket事件发生后通过回调函数内核会将其加入事件列表中用户调用epoll_wait() 返回文件描述符的个数不需要select/poll轮询整个socket集合
poll的服务态度要比select好很多，在通知我们的时候，不仅告诉我们有几个快递到了，还分别告诉我们是谁谁谁。我们只需要按照epoll给的答复，来询问快递员取快递即可。

while true {
    可处理的流[] = epoll_wait(epoll_fd); //阻塞

  //有消息抵达，全部放在 “可处理的流[]”中
    for i in 可处理的流[] {
        读 或者 其他处理
    }
}

接口API：

int epoll_create(int size);

epoll_ctl(epfd, EPOLL_CTL_ADD, 5, &new_event);

nt epoll_wait(int epfd, struct epoll_event *event,
int maxevents, int timeout);

示例：

int epfd = epoll_crete(1000);

//将 listen_fd 添加进 epoll 中
epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd,&listen_event);

while (1) {
    //阻塞等待 epoll 中 的fd 触发
    int active_cnt = epoll_wait(epfd, events, 1000, -1);

    for (i = 0 ; i < active_cnt; i++) {
        if (evnets[i].data.fd == listen_fd) {
            //accept. 并且将新accept 的fd 加进epoll中.
        }
        else if (events[i].events & EPOLLIN) {
            //对此fd 进行读操作
        }
        else if (events[i].events & EPOLLOUT) {
            //对此fd 进行写操作
        }
    }
}

水平触发：

如果用户在监听epoll事件，当内核有事件的时候，会拷贝给用户态事件，但是如果用户只处理了一次，那么剩下没有处理的会在下一次epoll_wait再次返回该事件

边缘触发：

相对跟水平触发相反，当内核有事件到达，只会通知用户一次，至于用户处理还是不处理，以后将不会再通知。这样减少了拷贝过程，增加了性能，但是相对来说，如果用户马虎忘记处理，将会产生事件丢的情况。

简单的epoll服务器：

// 服务端
#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <string.h>

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>

#include <sys/epoll.h>

#define SERVER_PORT (7778)
#define EPOLL_MAX_NUM (2048)
#define BUFFER_MAX_LEN (4096)

char buffer[BUFFER_MAX_LEN];

void str_toupper(char *str)
{
    int i;
    for (i = 0; i < strlen(str); i ++) {
        str[i] = toupper(str[i]);
    }
}

int main(int argc, char **argv)
{
    int listen_fd = 0;
    int client_fd = 0;
    struct sockaddr_in server_addr;
    struct sockaddr_in client_addr;
    socklen_t client_len;

    int epfd = 0;
    struct epoll_event event, *my_events;

    / socket
        listen_fd = socket(AF_INET, SOCK_STREAM, 0);

    // bind
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    server_addr.sin_port = htons(SERVER_PORT);
    bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr));

    // listen
    listen(listen_fd, 10);

    // epoll create
    epfd = epoll_create(EPOLL_MAX_NUM);
    if (epfd < 0) {
        perror("epoll create");
        goto END;
    }

    // listen_fd -> epoll
    event.events = EPOLLIN;
    event.data.fd = listen_fd;
    if (epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event) < 0) {
        perror("epoll ctl add listen_fd ");
        goto END;
    }

    my_events = malloc(sizeof(struct epoll_event) * EPOLL_MAX_NUM);


    while (1) {
        // epoll wait
        int active_fds_cnt = epoll_wait(epfd, my_events, EPOLL_MAX_NUM, -1);
        int i = 0;
        for (i = 0; i < active_fds_cnt; i++) {
            // if fd == listen_fd
            if (my_events[i].data.fd == listen_fd) {
                //accept
                client_fd = accept(listen_fd, (struct sockaddr*)&client_addr, &client_len);
                if (client_fd < 0) {
                    perror("accept");
                    continue;
                }

                char ip[20];
                printf("new connection[%s:%d]\n", inet_ntop(AF_INET, &client_addr.sin_addr, ip, sizeof(ip)), ntohs(client_addr.sin_port));

                event.events = EPOLLIN | EPOLLET;
                event.data.fd = client_fd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &event);
            }
            else if (my_events[i].events & EPOLLIN) {
                printf("EPOLLIN\n");
                client_fd = my_events[i].data.fd;

                // do read

                buffer[0] = '\0';
                int n = read(client_fd, buffer, 5);
                if (n < 0) {
                    perror("read");
                    continue;
                }
                else if (n == 0) {
                    epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &event);
                    close(client_fd);
                }
                else {
                    printf("[read]: %s\n", buffer);
                    buffer[n] = '\0';
#if 1
                    str_toupper(buffer);
                    write(client_fd, buffer, strlen(buffer));
                    printf("[write]: %s\n", buffer);
                    memset(buffer, 0, BUFFER_MAX_LEN);
#endif

                    /*
                       event.events = EPOLLOUT;
                       event.data.fd = client_fd;
                       epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);
                       */
                }
            }
            else if (my_events[i].events & EPOLLOUT) {
                printf("EPOLLOUT\n");
                /*
                   client_fd = my_events[i].data.fd;
                   str_toupper(buffer);
                   write(client_fd, buffer, strlen(buffer));
                   printf("[write]: %s\n", buffer);
                   memset(buffer, 0, BUFFER_MAX_LEN);

                   event.events = EPOLLIN;
                   event.data.fd = client_fd;
                   epoll_ctl(epfd, EPOLL_CTL_MOD, client_fd, &event);
                   */
            }
        }
    }

END:
    close(epfd);
    close(listen_fd);
    return 0;
}

// 客户端
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>

#define MAX_LINE (1024)
#define SERVER_PORT (7778)

void setnoblocking(int fd)
{
    int opts = 0;
    opts = fcntl(fd, F_GETFL);
    opts = opts | O_NONBLOCK;
    fcntl(fd, F_SETFL);
}

int main(int argc, char **argv)
{
    int sockfd;
    char recvline[MAX_LINE + 1] = {0};

    struct sockaddr_in server_addr;

    if (argc != 2) {
        fprintf(stderr, "usage ./client <SERVER_IP>\n");
        exit(0);
    }


    // 创建socket
    if ( (sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        fprintf(stderr, "socket error");
        exit(0);
    }


    // server addr 赋值
    bzero(&server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(SERVER_PORT);

    if (inet_pton(AF_INET, argv[1], &server_addr.sin_addr) <= 0) {
        fprintf(stderr, "inet_pton error for %s", argv[1]);
        exit(0);
    }


    // 链接服务端
    if (connect(sockfd, (struct sockaddr*) &server_addr, sizeof(server_addr)) < 0) {
        perror("connect");
        fprintf(stderr, "connect error\n");
        exit(0);
    }

    setnoblocking(sockfd);

    char input[100];
    int n = 0;
    int count = 0;



    // 不断的从标准输入字符串
    while (fgets(input, 100, stdin) != NULL)
    {
        printf("[send] %s\n", input);
        n = 0;
        // 把输入的字符串发送 到 服务器中去
        n = send(sockfd, input, strlen(input), 0);
        if (n < 0) {
            perror("send");
        }

        n = 0;
        count = 0;


        // 读取 服务器返回的数据
        while (1)
        {
            n = read(sockfd, recvline + count, MAX_LINE);
            if (n == MAX_LINE)
            {
                count += n;
                continue;
            }
            else if (n < 0){
                perror("recv");
                break;
            }
            else {
                count += n;
                recvline[count] = '\0';
                printf("[recv] %s\n", recvline);
                break;
            }
        }
    }

    return 0;
}

Reactor模型

重要组件：Event事件、Reactor反应堆、Demultiplex事件分发器、Evanthandler事件处理器

参考博文：https://xiaolincoding.com/os/8_network_system/reactor.html

组件	作用	对应代码（参考你的 `EchoServer`）
Event	事件抽象（如新连接、数据到达）	`TcpConnection` 的读写事件
Reactor	事件注册与核心调度器	`EventLoop` + `TcpServer`
Demultiplex	系统级事件监听（如 `epoll`）	`Poller`（底层调用 `epoll_wait`）
EventHandler	事件的具体处理逻辑	`EchoServer::onMessage` 等回调函数

流程图步骤：

2. 关键设计思想

(1) 事件驱动-Driven）

优势：避免轮询，CPU 仅在事件发生时被占用（高并发场景高效）。
你的代码体现：loop.loop() 是事件循环的核心，阻塞在 epoll_wait 直到事件发生。

(2) 非阻塞 I/O

Reactor 通常配合非阻塞 Socket（如 conn->shutdown() 直接操作，不阻塞线程）。

(3) 职责分离

Reactor 负责调度，EventHandler 负责业务逻辑（如 EchoServer 只关注 onMessage 的回显逻辑）。

muduo网络库的核心代码模块

Channels类：

channel类管理整个文件描述符，在TCP编程中想要IO多路复用监听某个文件描述符，需要把这个fd和对这个fd感兴趣的事件通过epoll_create 注册到多路复用模块（时间监听器），事件监听器监听该fd发生了某个事件，时间监听器返回发生fd的集合，以及每个fd发生了什么事件

EventLoop *loop_; // 事件循环,这个fd属于哪个EventLoop对象

const int fd_; // fd，Poller监听的对象 channel 对象照看的文件描述符

int events_; // 注册fd感兴趣的事件集合

int revents_;

Poller返回的具体发生的事件，代表事件监听器实际监听到该fd
发生的事件类型集合，当事件监听器监听到一个fd发生了什么事件，通过
Channel::set_revents()函数来设置revents值。

read_callback_、write_callback_、close_callback_、//error_callback_：这些是std::function类型，代表着这个Channel为这个文件描述符保存的各事件类型发生时的处理函数。比如这个fd发生了可读事件，需要执行可读事件处理函数，这时候Channel类都替你保管好了这些可调用函数，

Channel类的主要方法

void setReadCallback(ReadEventCallback cb) {read_callback_ = std::move(cb);}
void setWriteCallback(Eventcallback cb) {write_callback_ = std::move(cb);}
void setCloseCallback(EventCallback cb) {close_callback_ = std::move(cb);}
void setErrorCallback(EventCallback cb) {error_callback_ = std::move(cb);}

一个文件描述符会发生可读、可写、关闭、错误事件。当发生这些事件后，就需要调用相应的处理函数来处理。

将Channel中的文件描述符及其感兴趣事件注册事件监听器上或从事件监听器上移除

void enableReading() {events_ |= kReadEvent; upadte();}
void disableReading() {events_ &= ~kReadEvent; update();}
void enableWriting() {events_ |= kWriteEvent; update();}
void disableWriting() {events_ &= ~kWriteEvent; update();}
void disableAll() {events_ |= kNonEvent; update();}

外部通过这几个函数来告知Channel你所监管的文件描述符都对哪些事件类型感兴趣，并把这个文件描述符及其感兴趣事件注册到事件监听器（IO多路复用模块）上。这些函数里面都有一个update()私有成员方法，这个update其实本质上就是调用了epoll_ctl()。

int set_revents(int revt) {revents_ = revt;}
当事件监听器监听到某个文件描述符发生了什么事件，通过这个函数可以将这个文件描述符实际发生的事件封装进这个Channel中。

void HandlerEvent(TimeStamp receive_time)

当调用epoll_wait()后，可以得知事件监听器上哪些Channel（文件描述符）发生了哪些事件，事件发生后自然就要调用这些Channel对应的处理函数。 Channel::HandleEvent，让每个发生了事件的Channel调用自己保管的事件处理函数。每个Channel会根据自己文件描述符实际发生的事件（通过Channel中的revents_变量得知）和感兴趣的事件（通过Channel中的events_变量得知）来选择调用read_callback_和/或write_callback_和/或close_callback_和/或error_callback_。

Poller和EPollPoller - Demultiplex
Poller:负责监听文件描述符事件是否触发，以及返回发生事件的文件描述符，以及具体事件，一个Poller对应一个事件监听器，

epollfd_: 就是用epoll_create方法返回的epoll句柄
channels_：这个变量是std::unordered_map<int, Channel*>类型，负责记录文件描述符 —> Channel的映射，也帮忙保管所有- 注册在你这个Poller上的Channel。
ownerLoop_：所属的EventLoop对象

std::unordered_map<int, Channel*> channels

Timestamp poll(int timeoutMs, ChannelList* activeChannels) override;

数可以说是Poller的核心了，当外部调用poll方法的时候，该方法底层其实是通过epoll_wait获取这个事件监听器上发生事件的fd及其对应发生的事件，我们知道每个fd都是由一个Channel封装的，通过哈希表channels_可以根据fd找到封装这个fd的Channel。将事件监听器监听到该fd发生的事件写进这个Channel中的revents成员变量中。然后把这个Channel装进activeChannels中（它是一个vector<Channel*>）。这样，当外界调用完poll之后就能拿到事件监听器的监听结果（activeChannels_），

poll（）->::epoll_wait()

EventLoop - Reactor

需要循环的去【调用Poller:poll方法获取实际发生事件的Channel集合，然后调用这些Channel里面保管的不同类型事件的处理函数（调用Channel::HandlerEvent方法）

EventLoop就是负责实现“循环”，负责驱动“循环”的重要模块！！Channel和Poller其实相当于EventLoop的手下，EventLoop整合封装了二者并向上提供了更方便的接口来使用。

ChannelList activeChannels_;

std::unique_ptr poller_;

int wakeupFd; -> loop

std::unique_ptr wakeupChannel ;
(eventloop)

DeepSeek总结：
注册事件：
用户创建一个 Channel 并绑定到某个 fd（如 socket）。
调用 Channel::enableReading() 等方法，通过 EventLoop 向 EventPoller 注册关注的事件。

事件监听：
EventLoop 调用 EventPoller::poll() 阻塞等待事件。
当 fd 事件就绪时，EventPoller 返回活跃的 Channel 列表。

事件处理：
EventLoop 遍历活跃的 Channel，调用它们的 handleEvent() 方法，触发预先注册的回调函数（如 ReadCallback）。

任务调度：
EventLoop 还可以处理非 I/O 任务（如定时任务或跨线程任务），通过 runInLoop 机制确保线程安全。

EventLoop (驱动循环)
└── EventPoller (监听事件)
└── Channel列表 (处理具体事件)

Thread和EventLoopThread
每一个EventLoop都绑定了一个线程（一对一绑定)
EventLoopThreadPool

getNextLoop() : 通过轮询算法获取下一个subloop baseLoop

一个thread对应一个loop => one loop per thread

Socket

int fd() const { return sockfd_; }
  // return true if success.
  bool getTcpInfo(struct tcp_info*) const;
  bool getTcpInfoString(char* buf, int len) const;

  /// abort if address in use
  void bindAddress(const InetAddress& localaddr);
  /// abort if address in use
  void listen();

  /// On success, returns a non-negative integer that is
  /// a descriptor for the accepted socket, which has been
  /// set to non-blocking and close-on-exec. *peeraddr is assigned.
  /// On error, -1 is returned, and *peeraddr is untouched.
  int accept(InetAddress* peeraddr);

  void shutdownWrite();

  ///
  /// Enable/disable TCP_NODELAY (disable/enable Nagle's algorithm).
  ///
  void setTcpNoDelay(bool on);

  ///
  /// Enable/disable SO_REUSEADDR
  ///
  void setReuseAddr(bool on);

  ///
  /// Enable/disable SO_REUSEPORT
  ///
  void setReusePort(bool on);

  ///
  /// Enable/disable SO_KEEPALIVE
  ///
  void setKeepAlive(bool on);

 private:
  const int sockfd_;

Acceptor

主要封装了listenfd相关的操作 socket bind listen baseLoop
Accetpor封装了服务器监听套接字fd以及相关处理方法。Acceptor类内部其实没有贡献什么核心的处理函数，主要是对其他类的方法调用进行封装。

acceptSocket_：这个是服务器监听套接字的文件描述符

acceptChannel_：这是个Channel类，把acceptSocket_及其感兴趣事件和事件对应的处理函数都封装进去。

EventLoop *loop：监听套接字的fd由哪个EventLoop负责循环监听以及处理相应事件，其实这个EventLoop就是main EventLoop。

newConnectionCallback_: TcpServer构造函数中将- TcpServer::newConnection( )函数注册给了这个成员变量。这个- TcpServer::newConnection函数的功能是公平的选择一个subEventLoop，并把已经接受的连接分发给这个subEventLoop。

方法：
listen( )：该函数底层调用了linux的函数listen( )，开启对acceptSocket_的监听同时将acceptChannel及其感兴趣事件（可读事件）注册到main EventLoop的事件监听器上。换言之就是让main EventLoop事件监听器去监听acceptSocket_

handleRead( )：这是一个私有成员方法，这个方法是要注册到acceptChannel_上的，同时handleRead( )方法内部还调用了成员变量newConnectionCallback_保存的函数。当main EventLoop监听到acceptChannel_上发生了可读事件时（新用户连接事件），就是调用这个handleRead( )方法。

简单说一下这个handleRead( )最终实现的功能是什么，接受新连接，并且以负载均衡的选择方式选择一个sub EventLoop，并把这个新连接分发到这个subEventLoop上。

Buffer

缓冲区应用写数据 -> 缓冲区 -> Tcp发送缓冲区 -> send

prependable readeridx writeridx

TcpConnection

一个连接成功的客户端对应一个TcpConnection Socket Channel 各种回调发送和接收缓冲

区TcpServer

Acceptor EventLoopThreadPool

ConnectionMap connections_;

代码：https://github.com/Charlie-todream/c11moduuo

本文作者： 东方觉主
本文链接： http://www.charon193.com/2025/04/01/mymoduo/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！