网络层设计:主从 Reactor + One Loop Per Thread

代码:src/network/ 8 个核心类,共 ~800 行灵感:陈硕《Linux 多线程服务端编程》+ Muduo 源码 设计目标:0 锁的 IO 路径(每连接绑定一个 EventLoop,无共享)

写 RPC 框架,网络层是最值得"自己写一遍"的部分——HTTP 框架、RPC、消息推送、游戏服务器,网络层都是这套。

我对照陈硕《Linux 多线程服务端编程》自己撸了一遍,没抄 Muduo 源码——边看边写,踩了 6 个坑,全部记下来。

一、整体架构

主从 Reactor + One Loop Per Thread:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
┌─────────────────────────────────────────────────────────────┐
│                        TcpServer                             │
│  ┌─────────────┐    ┌─────────────────────────────────────┐  │
│  │  Acceptor   │───▶│      EventLoopThreadPool           │  │
│  │ (主 Reactor)│    │  ┌─────────┐ ┌─────────┐ ┌──────┐  │  │
│  │  listen fd  │    │  │IO Loop 0│ │IO Loop 1│ │ ...  │  │  │
│  │  监听新连接  │    │  │(sub)    │ │(sub)    │ │      │  │  │
│  └─────────────┘    │  └─────────┘ └─────────┘ └──────┘  │  │
│         │           └─────────────────────────────────────┘  │
│         │                      ▲                             │
│         └──────────────────────┘                             │
│                    轮询分发新连接                              │
└─────────────────────────────────────────────────────────────┘

核心流程:

主 Reactor(baseLoop)跑 Acceptor,监听 listen_fd
新连接 accept() → getNextLoop() 轮询分到 subLoop
每个 TcpConnection 绑定一个 subLoop,该 Loop 负责该连接所有 IO
One Loop Per Thread:每个 Loop 跑一个线程,无锁处理读写

二、8 大核心类

2.1 Socket —— fd 的 RAII

目的:管理 fd 生命周期,避免 double close / fd 泄漏。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
class Socket {
 public:
  explicit Socket(int sockfd) : sockfd_(sockfd) {}
  ~Socket() { ::close(sockfd_); }
  Socket(const Socket&) = delete;
  Socket& operator=(const Socket&) = delete;

  void bindAddress(const sockaddr_in& localaddr);
  void listen();
  int accept(sockaddr_in* peeraddr);
  void shutdownWrite();

  void setTcpNoDelay(bool on);    // 禁 Nagle
  void setReuseAddr(bool on);
  void setReusePort(bool on);
  void setKeepAlive(bool on);

 private:
  int sockfd_;
};

关键点:

explicit 禁止隐式转换(int 不能误转 Socket)
删拷贝构造 + 赋值(fd 唯一)
accept4 替代 accept + fcntl:accept4(SOCK_NONBLOCK | SOCK_CLOEXEC) 一次系统调用同时设非阻塞 + 关闭时自动关闭 fd

accept4 是关键优化——传统 accept() + fcntl(O_NONBLOCK) 是两次系统调用,高并发下影响 QPS。

2.2 Buffer —— 应用层缓冲区

目的:处理 TCP 粘包/半包,支持 prepend(协议头放在数据前)。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class Buffer {
 public:
  static const size_t kCheapPrepend = 8;
  static const size_t kInitialSize = 1024;
  
  size_t readableBytes() const;
  size_t writableBytes() const;
  size_t prependableBytes() const;
  
  void append(const char* data, size_t len);
  void retrieve(size_t len);
  std::string retrieveAllAsString();
  
  ssize_t readFd(int fd, int* savedErrno);
  ssize_t writeFd(int fd, int* savedErrno);

 private:
  std::vector<char> buffer_;
  size_t reader_index_ = kCheapPrepend;
  size_t writer_index_ = kCheapPrepend;
};

内存布局:

1
2
3
4
5
6
7
8
┌─────────────────┬──────────────────┬─────────────────┐
│  prependable    │    readable      │    writable     │
│  (已读空间)      │   (未读数据)      │   (空闲空间)     │
│                 │                  │                 │
│  0 ──readerIndex_── writerIndex_ ─── buffer_.size() │
│  ↑ 头部预留 8B  │                  │                 │
│  协议头可塞这里  │  peek()/retrieve │  append()       │
└─────────────────┴──────────────────┴─────────────────┘

readFd 用 readv 优化:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
ssize_t Buffer::readFd(int fd, int* savedErrno) {
  char extrabuf[65536];  // 栈上 64KB
  struct iovec vec[2];
  vec[0].iov_base = begin() + writer_index_;
  vec[0].iov_len = writableBytes();
  vec[1].iov_base = extrabuf;
  vec[1].iov_len = sizeof(extrabuf);
  
  ssize_t n = readv(fd, vec, 2);
  if (n < 0) {
    *savedErrno = errno;
  } else if (n <= writableBytes()) {
    writer_index_ += n;
  } else {
    writer_index_ = buffer_.size();
    append(extrabuf, n - writableBytes());
  }
  return n;
}

readv 妙处:

缓冲区有空间 → 直接读进缓冲区
缓冲区满 → 读进栈上 extrabuf,避免反复 read
一次 readv 处理"缓冲区满 + 栈临时"两个目标,数据只拷一次

2.3 Channel —— fd 事件分发

目的:绑定一个 fd,注册事件,事件触发时调回调。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Channel {
 public:
  void handleEvent();           // 事件分发入口
  void enableReading();
  void disableReading();
  void enableWriting();
  void disableWriting();
  void disableAll();
  void remove();
  
  void setReadCallback(EventCallback cb);
  void setWriteCallback(EventCallback cb);
  void setErrorCallback(EventCallback cb);
  void setCloseCallback(EventCallback cb);

 private:
  static const int kNew = -1;
  static const int kAdded = 1;
  static const int kDeleted = 2;
  
  EventLoop* loop_;
  const int fd_;
  int events_;
  int revents_;
  int index_;  // -1 / 1 / 2
  EventCallback readCallback_;
  EventCallback writeCallback_;
  EventCallback errorCallback_;
  EventCallback closeCallback_;
};

事件处理顺序(我栽过的坑):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
void Channel::handleEvent() {
  // 1. HUP 但有数据可读 → 优先读(不丢数据)
  if ((revents_ & EPOLLHUP) && !(revents_ & EPOLLIN)) {
    closeCallback_();
  }
  // 2. 错误
  if (revents_ & EPOLLERR) {
    errorCallback_();
  }
  // 3. 可读(含 EPOLLRDHUP 半关闭)
  if (revents_ & (EPOLLIN | EPOLLPRI | EPOLLRDHUP)) {
    readCallback_();
  }
  // 4. 可写
  if (revents_ & EPOLLOUT) {
    writeCallback_();
  }
}

关键:EPOLLHUP 不一定要关——对端关写但本端还有数据要读(EPOLLIN 同时),要先把数据读完。

2.4 EventLoop —— 事件循环核心

目的:epoll_wait + 定时器 + 跨线程任务投递。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class EventLoop {
 public:
  void loop();              // 主循环
  void quit();              // 退出
  
  void updateChannel(Channel*);
  void removeChannel(Channel*);
  
  void runInLoop(Func cb);    // Loop 线程:直接跑;其他线程:投递
  void queueInLoop(Func cb);
  
  void runEvery(double sec, Func cb);  // 周期定时器
  void runAfter(double sec, Func cb);  // 一次性

 private:
  void handleRead();        // wakeup fd 读
  void doPendingFunctors();
  
  std::unique_ptr<EpollPoller> poller_;
  int wakeup_fd_;           // eventfd,跨线程唤醒
  Channel wakeup_channel_;
  std::vector<Func> pending_functors_;
  std::mutex mutex_;
};

loop() 主循环:

1
2
3
4
5
6
7
8
9
void EventLoop::loop() {
  while (!quit_) {
    poller_->poll(10000, &active_channels_);  // 10s timeout
    for (auto* ch : active_channels_) {
      ch->handleEvent();
    }
    doPendingFunctors();
  }
}

跨线程唤醒:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
其他线程 queueInLoop(cb)
    │
    ▼
加锁 → push cb 到 pending_functors_
    │
    ▼
wakeup() → write(eventfd, 1)  // 8 字节
    │
    ▼
EventLoop 线程被 epoll_wait 唤醒
    │
    ▼
doPendingFunctors() 加锁 → swap(队列) → 锁外执行 cb

swap 而不是直接遍历:锁内只 swap 指针,O(1);锁外执行回调(可能耗时)。

2.5 EventLoopThread —— IO 线程封装

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
class EventLoopThread {
 public:
  EventLoop* startLoop();   // 启动线程,返回 Loop*
  void stop();

 private:
  void threadFunc();         // 线程函数:建 Loop → 跑

  EventLoop* loop_;
  std::thread thread_;
  std::mutex mutex_;
  std::condition_variable cond_;
  ThreadInitCallback callback_;
  bool started_;
  bool quitting_;
};

startLoop() 阻塞等线程就绪——用 cond_var 通知,确保调用方拿到 Loop 时它已经在跑。

2.6 EventLoopThreadPool —— IO 线程池

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
class EventLoopThreadPool {
 public:
  void setThreadNum(int n) { num_threads_ = n; }
  void start(const ThreadInitCallback& cb);
  
  EventLoop* getNextLoop();           // 轮询
  EventLoop* getLoopForHash(size_t);  // 一致性哈希

 private:
  EventLoop* base_loop_;             // 主 Reactor 的 Loop
  std::vector<std::unique_ptr<EventLoopThread>> threads_;
  std::vector<EventLoop*> loops_;
  int num_threads_;
  int next_;  // 轮询索引
};

getNextLoop() 必须 baseLoop 线程调(有 assertInLoopThread),否则多个 Acceptor 抢 next_ 索引。

2.7 Acceptor —— 监听新连接

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class Acceptor {
 public:
  Acceptor(EventLoop* loop, const sockaddr_in& listen_addr);
  void listen();  // bind + listen + 注册到 baseLoop

 private:
  void handleRead();  // accept 新连接
  
  EventLoop* loop_;
  Socket accept_socket_;
  Channel accept_channel_;
  NewConnectionCallback new_connection_cb_;
  bool listening_;
};

半同步/半异步:

异步:baseLoop epoll 监听 listen_fd 可读 → 调 handleRead() → accept4()
同步:accept4() 拿到 connfd 后,同步回调 new_connection_cb_(connfd, peer_addr) → 由 TcpServer 决定分到哪个 subLoop

2.8 TcpConnection —— 连接生命周期

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class TcpConnection : public enable_shared_from_this<TcpConnection> {
 public:
  void send(Buffer& buf);
  void send(const string& msg);
  void shutdown();
  void forceClose();
  
  void setConnectionCallback(Callback cb);
  void setMessageCallback(Callback cb);
  void setWriteCompleteCallback(Callback cb);
  void setCloseCallback(Callback cb);

 private:
  enum State { kDisconnected, kConnecting, kConnected, kDisconnecting };
  
  EventLoop* loop_;
  unique_ptr<Socket> socket_;
  unique_ptr<Channel> channel_;
  Buffer input_buffer_;
  Buffer output_buffer_;
  State state_;
};

enable_shared_from_this——TcpConnection 由 TcpServer / Channel / 应用多处引用,shared_ptr 统一管理,避免"Channel 还在用,connection 已析构"。

send 不是直接 write:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
void TcpConnection::send(Buffer& buf) {
  if (state_ == kConnected) {
    if (loop_->isInLoopThread()) {
      sendInLoop(buf);
    } else {
      loop_->runInLoop([this, buf] { sendInLoop(buf); });
    }
  }
}

void TcpConnection::sendInLoop(Buffer& buf) {
  ssize_t n = ::write(fd_, buf.peek(), buf.readableBytes());
  if (n < writableBytes()) {
    // 没写完,剩下的塞 output_buffer_,等 EPOLLOUT
    output_buffer_.append(buf.peek() + n, buf.readableBytes() - n);
    channel_->enableWriting();
  }
}

非阻塞 IO 关键:

write 返回 EAGAIN → 塞 output_buffer_,注册 EPOLLOUT,等可写再发
EPOLLOUT 触发 → 继续 write,直到 output_buffer_ 空 → 注销 EPOLLOUT

三、3 个关键线程模型原则

3.1 One Loop Per Thread

每 IO 线程一个 EventLoop,无锁处理读写——所有 IO 都在 Loop 线程,不跨线程。

反例:多线程共享一个 EventLoop → 多线程 epoll_wait 同一个 epoll fd → 同一 fd 被多个线程处理,race condition。

3.2 跨线程必须 `runInLoop`

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
// 错
void TcpConnection::send(Buffer& buf) {
  ::write(fd_, buf.peek(), buf.readableBytes());  // 别的线程调,fd 不在当前 Loop
}

// 对
void TcpConnection::send(Buffer& buf) {
  if (loop_->isInLoopThread()) {
    sendInLoop(buf);
  } else {
    loop_->runInLoop([this, buf] { sendInLoop(buf); });
  }
}

isInLoopThread 判断——是本线程就直接执行,不是就投递。

3.3 析构必须 `~EventLoopThread` 在 quit 后

1
2
3
4
5
6
7
8
~EventLoopThread() {
  if (loop_) {
    loop_->quit();
  }
  if (thread_.joinable()) {
    thread_.join();
  }
}

先 quit loop → 再 join 线程——子线程 loop() 退出后,主线程才能 join 收尸。

四、踩过的 6 个坑

4.1 `accept` 没设 `SOCK_CLOEXEC` → fd 泄漏

fork + exec 后 fd 还在,浪费资源。accept4(SOCK_NONBLOCK | SOCK_CLOEXEC) 一次解决。

4.2 `EPOLLHUP` 当成 close,丢数据

EPOLLHUP + EPOLLIN 同时 → 先 EPOLLIN 读,再处理关闭。我栽过:对端关写但还有 1KB 数据,被我直接关连接,数据丢。

4.3 `Buffer::readFd` 不用 readv → 反复 read

read() 一次只读 N 字节,缓冲区小就反复 read 系统调用。readv + 栈临时缓冲,一次 read 完。

4.4 `Channel::remove` 没注销 epoll → fd 泄漏

Channel 析构前必须 disableAll() + remove(),否则 epoll 还持有 fd,fd 永远不释放。

4.5 EventLoop 析构前没 quit → 线程死循环

EventLoop 析构时 loop_ 还在跑,thread_.join() 永远不返回——死锁。

4.6 `mutex` 锁内调回调 → 死锁

doPendingFunctors 锁内不能调回调——回调可能又 queueInLoop,锁嵌套死锁。锁内 swap 队列,锁外执行回调。

五、性能数据(本机回环,4 核)

场景	QPS	p99
单连接 1 字节 echo	250K	80μs
单连接 1KB echo	80K	220μs
1000 并发连接	35K	380μs
1000 连接 + 5s 长连接	60K	280μs

核心:One Loop Per Thread,IO 路径 0 锁。

六、上层封装:TcpServer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
class TcpServer {
 public:
  TcpServer(EventLoop* loop, const sockaddr_in& addr, string name);
  void start();  // 启动 Acceptor + IO 线程池
  
  void setMessageCallback(MessageCallback cb);
  void setConnectionCallback(ConnectionCallback cb);

 private:
  void newConnection(int sockfd, const sockaddr_in& peer);
  void removeConnection(const TcpConnectionPtr& conn);
  
  EventLoop* loop_;  // baseLoop
  unique_ptr<Acceptor> acceptor_;
  shared_ptr<EventLoopThreadPool> thread_pool_;
  atomic<int> started_;
  int next_conn_id_;
  ConnectionMap connections_;  // name → TcpConnection
};

newConnection:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
void TcpServer::newConnection(int sockfd, const sockaddr_in& peer) {
  EventLoop* io_loop = thread_pool_->getNextLoop();
  string name = name_ + "-" + to_string(next_conn_id_++);
  auto conn = make_shared<TcpConnection>(io_loop, name, sockfd);
  connections_[name] = conn;
  conn->setMessageCallback(message_cb_);
  conn->setConnectionCallback(connection_cb_);
  conn->setCloseCallback([this](auto c) { removeConnection(c); });
  io_loop->runInLoop([conn] { conn->connectEstablished(); });
}

关键:runInLoop 触发 connectEstablished——把 TcpConnection 的状态从 kConnecting 转到 kConnected,注册到 epoll。

七、推荐阅读

陈硕《Linux 多线程服务端编程》——讲 EventLoop 最清楚
Muduo 源码(github.com/chenshuo/muduo)——生产级 C++ 网络库
《UNIX 网络编程》卷 1——socket / epoll 基础

自己写一遍网络层,是 C++ 后端的分水岭——看完陈硕的书,再写一个类 Muduo 库,网络编程就真的懂了。

相关阅读: