解析muduo源码之 Acceptor.h & Acceptor.cc

// Copyright 2010, Shuo Chen. 保留所有权利。
// http://code.google.com/p/muduo/
//
// 本源代码的使用受 BSD 许可证约束
// 可在 License 文件中找到许可证条款。

// 作者：陈硕 (chenshuo at chenshuo dot com)
//
// 这是一个内部头文件，你不应该直接包含它。

#ifndef MUDUO_NET_ACCEPTOR_H
#define MUDUO_NET_ACCEPTOR_H

#include <functional>

#include "muduo/net/Channel.h"
#include "muduo/net/Socket.h"

namespace muduo
{
namespace net
{

class EventLoop;
class InetAddress;

///
/// TCP 连接接收器：负责监听端口、接受新的客户端连接
///
/// 核心作用：监听 socket 产生可读事件时，调用 accept() 接受连接
/// 并将新连接的 sockfd 和对端地址通过回调交给 TcpServer 处理
class Acceptor : noncopyable
{
 public:
  /// 新连接回调类型
  /// 参数：新连接的 sockfd、对端地址
  typedef std::function<void (int sockfd, const InetAddress&)> NewConnectionCallback;

  /// 构造函数
  /// @param loop 所属的事件循环（IO 线程）
  /// @param listenAddr 要监听的本地地址（IP + 端口）
  /// @param reuseport 是否开启端口复用（SO_REUSEPORT）
  Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport);
  ~Acceptor();

  /// 设置新连接到来时的回调函数（由 TcpServer 设置）
  void setNewConnectionCallback(const NewConnectionCallback& cb)
  { newConnectionCallback_ = cb; }

  /// 开始监听端口
  void listen();

  /// 返回当前是否正在监听
  bool listening() const { return listening_; }

  // 已废弃，拼写错误，仅保留用于错误搜索
  // bool listenning() const { return listening(); }

 private:
  /// 处理监听 socket 的可读事件（有新连接到来）
  void handleRead();

  EventLoop* loop_;              // 所属的事件循环
  Socket acceptSocket_;          // 监听用的 socket（服务端 socket）
  Channel acceptChannel_;        // 管理监听 socket 事件的通道
  NewConnectionCallback newConnectionCallback_;  // 新连接回调
  bool listening_;               // 是否正在监听
  int idleFd_;                   // 占位空闲 fd，用于处理文件描述符耗尽的情况
};

}  // namespace net
}  // namespace muduo

#endif  // MUDUO_NET_ACCEPTOR_H

1. 整体定位

Acceptor = 服务器监听套接字的专属管理者

只做一件事：监听端口、接收新连接
运行在 main Reactor（baseLoop）中
接收连接后，回调给 TcpServer
内部封装 listen socket + channel

2. 核心作用

创建、绑定、监听 socket
当有新连接到来时，触发 handleRead()
调用 accept() 拿到新连接的 connfd
调用回调函数 newConnectionCallback_(connfd, ...) 交给 TcpServer

3. 类结构与关键成员

class Acceptor : noncopyable
{
public:
  typedef std::function<void(int sockfd, const InetAddress&)> NewConnectionCallback;

  Acceptor(EventLoop* loop, ...);
  void listen();
  void setNewConnectionCallback(NewConnectionCallback cb);

private:
  void handleRead();  // 新连接到来时触发

  EventLoop* loop_;
  Socket acceptSocket_;    // 监听 socket（listen_fd）
  Channel acceptChannel_;  // 监听 socket 对应的 channel
  NewConnectionCallback newConnectionCallback_;
  bool listening_;
  int idleFd_;             // 占位 fd，处理文件描述符耗尽
};

4. 成员作用一览

1. loop_

所属的 EventLoop（main loop）
所有 accept 操作都在这个 IO 线程

2. acceptSocket_

监听套接字（server socket）
负责 bind() + listen()

3. acceptChannel_

监听套接字的事件分发器
当可读（有新连接）时调用 handleRead()

4. newConnectionCallback_

接收新连接后回调给 TcpServer
TcpServer 用它来创建 TcpConnection

5. idleFd_

一个占位用的 fd（通常打开 /dev/null）
当文件描述符耗尽时，先占用一个，避免惊群与死循环

5. 核心函数

1. 构造函数

Acceptor::Acceptor(EventLoop* loop, ...)
  : loop_(loop),
    acceptSocket_(...),  // 创建 listen socket
    acceptChannel_(loop, acceptSocket_.fd())
{
  acceptChannel_.setReadCallback(std::bind(&Acceptor::handleRead, this));
}

作用：

创建监听 socket
绑定到 channel
设置可读回调为 handleRead()

2. listen()

void Acceptor::listen()
{
  loop_->assertInLoopThread();
  acceptSocket_.listen();   // 系统调用 listen()
  listening_ = true;
  acceptChannel_.enableReading(); // 开始监听读事件
}

真正让服务器开始监听端口。

3. handleRead ()

void Acceptor::handleRead()
{
  int connfd = acceptSocket_.accept(&peerAddr);
  if (connfd >= 0)
  {
    if (newConnectionCallback_)
      newConnectionCallback_(connfd, peerAddr);
  }
  else
  {
    // 处理 fd 耗尽等错误
  }
}

这就是接收新连接的地方！

流程：

有客户端连接 → listen socket 可读
channel 触发 handleRead()
调用 accept() → 得到 connfd
调用回调 → 交给 TcpServer

6. Acceptor 在 Muduo 架构中的位置

TcpServer
  ↓
main EventLoop（baseLoop）
  ↓
Acceptor（负责 listen + accept）
  ↓
新连接 → 回调 → TcpServer → 创建 TcpConnection

Acceptor 只负责接收连接，不负责处理连接！

7. 为什么要有 idleFd_？

Linux 系统最大打开文件数有限制
如果 fd 耗尽，accept() 会失败
此时如果不处理，会一直触发可读事件，导致死循环
提前打开一个 idleFd_，出错时先关闭它，腾出一个 fd
处理完错误再恢复

这是 高并发服务必备的鲁棒性设计。

二、 Acceptor.cc

先贴出完整代码，再逐部分解释：

// Copyright 2010, Shuo Chen. 保留所有权利。
// http://code.google.com/p/muduo/
//
// 本源代码的使用受 BSD 许可证约束
// 可在 License 文件中找到许可证条款。

// 作者：陈硕 (chenshuo at chenshuo dot com)

#include "muduo/net/Acceptor.h"

#include "muduo/base/Logging.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/InetAddress.h"
#include "muduo/net/SocketsOps.h"

#include <errno.h>
#include <fcntl.h>
#include <unistd.h>

using namespace muduo;
using namespace muduo::net;

// 构造函数：创建监听 socket、绑定地址、初始化 channel
Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport)
  : loop_(loop),                                  // 绑定事件循环
    acceptSocket_(sockets::createNonblockingOrDie(listenAddr.family())), // 创建非阻塞监听 socket
    acceptChannel_(loop, acceptSocket_.fd()),     // 用 channel 管理监听 fd
    listening_(false),                            // 初始未监听
    idleFd_(::open("/dev/null", O_RDONLY | O_CLOEXEC))  // 打开一个空闲 fd，用于处理文件描述符耗尽
{
  assert(idleFd_ >= 0);
  acceptSocket_.setReuseAddr(true);        // 设置地址复用
  acceptSocket_.setReusePort(reuseport);   // 设置端口复用（可选）
  acceptSocket_.bindAddress(listenAddr);   // 绑定监听地址
  // 设置 channel 可读回调：有新连接时调用 handleRead
  acceptChannel_.setReadCallback(
      std::bind(&Acceptor::handleRead, this));
}

// 析构函数：关闭 channel、关闭空闲 fd
Acceptor::~Acceptor()
{
  acceptChannel_.disableAll();    // 取消所有事件监听
  acceptChannel_.remove();         // 从事件循环移除
  ::close(idleFd_);                // 关闭空闲文件描述符
}

// 开始监听端口
void Acceptor::listen()
{
  loop_->assertInLoopThread();     // 必须在 IO 线程调用
  listening_ = true;               // 标记为监听状态
  acceptSocket_.listen();          // 系统调用 listen
  acceptChannel_.enableReading();  // 开始监听可读事件（等待新连接）
}

// 处理新连接到达（核心函数）
void Acceptor::handleRead()
{
  loop_->assertInLoopThread();

  InetAddress peerAddr;  // 对端地址

  // 接受新连接
  int connfd = acceptSocket_.accept(&peerAddr);
  if (connfd >= 0)
  {
    // 成功拿到新连接的 fd
    if (newConnectionCallback_)
    {
      // 回调给 TcpServer 创建 TcpConnection
      newConnectionCallback_(connfd, peerAddr);
    }
    else
    {
      // 没有设置回调，直接关闭连接
      sockets::close(connfd);
    }
  }
  else
  {
    // accept 失败
    LOG_SYSERR << "in Acceptor::handleRead";

    // 处理文件描述符耗尽（EMFILE）的经典方案
    // 来自 libev 文档：进程打开 fd 达到上限，accept 失败
    if (errno == EMFILE)
    {
      // 1. 先关掉空闲 fd，腾出一个名额
      ::close(idleFd_);
      // 2. accept 拿走这个新连接（拿到 fd）
      idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL);
      // 3. 立刻关掉这个连接
      ::close(idleFd_);
      // 4. 重新打开 /dev/null 恢复空闲 fd
      idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC);
    }
  }
}

1. Acceptor 到底是干嘛的？

Acceptor 就是专门负责：

bind + listen 一个端口
等待客户端连接
调用 accept() 拿到新连接 fd
交给 TcpServer 创建 TcpConnection

它运行在 main EventLoop（主 Reactor）。

2. 构造函数 Acceptor::Acceptor ()

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport)
  : loop_(loop),
    acceptSocket_(sockets::createNonblockingOrDie(listenAddr.family())),
    acceptChannel_(loop, acceptSocket_.fd()),
    listening_(false),
    idleFd_(::open("/dev/null", O_RDONLY | O_CLOEXEC))
{
  assert(idleFd_ >= 0);
  acceptSocket_.setReuseAddr(true);
  acceptSocket_.setReusePort(reuseport);
  acceptSocket_.bindAddress(listenAddr);
  acceptChannel_.setReadCallback(
      std::bind(&Acceptor::handleRead, this));
}

逐行解释：

loop_绑定主 EventLoop，所有操作都在主线程。
acceptSocket_创建非阻塞监听 socket。
acceptChannel_把 listen fd 包装成 channel，由 EventLoop 监听事件。
idleFd_ = open("/dev/null")打开一个占位文件描述符，应对 EMFILE 错误（fd 耗尽）。
setReuseAddr / setReusePort端口复用，允许快速重启服务器。
bindAddress(listenAddr)绑定 IP:PORT。
setReadCallback(&Acceptor::handleRead)当有新连接时 → 调用 handleRead()。

3. 析构函数～Acceptor ()

Acceptor::~Acceptor()
{
  acceptChannel_.disableAll();   // 取消事件
  acceptChannel_.remove();       // 从loop中移除
  ::close(idleFd_);              // 关闭占位fd
}

4. listen () —— 开始监听端口

void Acceptor::listen()
{
  loop_->assertInLoopThread();
  listening_ = true;
  acceptSocket_.listen();        // 系统调用 listen()
  acceptChannel_.enableReading();// 开始监听可读事件
}

作用：调用 listen(fd, SOMAXCONN)让 socket 进入监听状态。

5. handleRead () —— 接收新连接

void Acceptor::handleRead()
{
  loop_->assertInLoopThread();
  InetAddress peerAddr;

  // 接受客户端连接
  int connfd = acceptSocket_.accept(&peerAddr);

  if (connfd >= 0)
  {
    // 有新连接！回调给 TcpServer
    if (newConnectionCallback_) {
      newConnectionCallback_(connfd, peerAddr);
    } else {
      sockets::close(connfd);
    }
  }
  else
  {
    // 出错！重点：处理 EMFILE
    if (errno == EMFILE)
    {
      ::close(idleFd_);                // 先关掉占位fd
      idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL); // 拿走一个连接
      ::close(idleFd_);                // 关掉它
      idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC); // 恢复占位fd
    }
  }
}

6. 这个函数到底做了什么？

正常流程：

客户端连接 → listenfd 可读
channel 触发 handleRead()
accept() → 得到 connfd
调用 newConnectionCallback_(connfd, peerAddr)
TcpServer 收到 connfd → 创建 TcpConnection

错误流程（EMFILE 文件描述符满）：

这是 高并发服务器必处理的错误：

进程打开 fd 达到上限
accept 失败，返回 EMFILE
但内核还会一直通知可读事件
导致死循环、100% CPU

Acceptor 的解决方案（经典设计）：

先关闭 idleFd_（腾出一个 fd）
调用 accept 拿走一个连接（得到一个 fd）
立刻关闭它（拒绝连接）
重新打开 /dev/null 恢复 idleFd_

这样就不会死循环，还能优雅拒绝过载连接。

7. Acceptor 在 Muduo 架构中的位置

TcpServer
   ↓
main EventLoop (baseLoop)
   ↓
Acceptor  ::  bind + listen + accept
   ↓
新连接 connfd
   ↓
TcpServer 创建 TcpConnection
   ↓
分配到 子 EventLoop（IO线程）

Acceptor = 主反应器的连接接收器

TcpConnection = 子反应器的连接处理器