先回顾下Tcp连接建立、数据通信、连接断开的过程。一个典型的TCP连接、通信过程：

from UNP第3版 P36

TCP连接中，两端是对等的，TCP协议也没有区分客户端（client）与服务器端（server），但互联网中通信中，往往有一端提供资源给另一端访问，我们把拥有资源的一端称为服务器端，请求资源的一端称为客户端。

对于server，会启用一个监听循环，不断接受client连接请求（三次握手），进行数据通信，通信完成以后断开连接（四次挥手）。

对于client，在server启用监听循环时，向server发出连接请求，接收server数据，如有必要向server发送数据，通信完成后，断开连接。

连接是指物理上一端（client）到另一端（server）的通信链路，通过server端<ip, port>与客户端<ip, port>，来唯一标识一个TCP连接。

TCP连接有长连接、短连接之分：
短连接：client和server建立连接后，一般只传递一次读写操作，然后由client发起close，发送FIN分节，关闭连接。短连接只完成一次read/write操作，就会自动关闭。
长连接：client和server建立连接后，并不会自动关闭，后续的read/write操作会继续用这个连接。长连接没有确切时间限制，可能会长时间存在。

本文主要从Tcp连接的建立与关闭，来分析源码。

TcpServer接受新连接

muduo使用TcpConnection类来管理TCP连接，使用接受器Acceptor来接受连接，连接器Connector发起连接。TcpServer管理accept获得TcpConnection，生命周期由用户控制。

下图是TcpServer新建连接的相关函数调用顺序。当Channel::handleEvent()的触发条件是listening socket可读时，表明有新连接请求达到。TcpServer为新连接创建对应的TcpConnection对象。

Acceptor类

Acceptor是TcpServer的一个内部类，主要职责是用来获得新连接的fd。保存用户提供的Connection-Callback和MessageCallback，新建TcpConnection对象（newConn()）的时候直接传递给TcpConnection的构造函数。

/**
* TCP连接接受器
* 基础调用为accept(2)/accept4(2)
*/
class Acceptor : private noncopyable
{
public:
    typedef std::function<void(int sockfd, const InetAddress &)> NewConnectionCallback;

    Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport);
    ~Acceptor();

    /* 设置新连接回调 */
    void setNewConnectionCallback(const NewConnectionCallback& cb)
    { newConnectionCallback_ = cb; }

    /* 监听本地端口 */
    void listen();

    /* 判断当前是否正在监听端口 */
    bool listening() const { return listening_; }

private:
    void handleRead();      // 处理读事件

    EventLoop *loop_;        // 所属EventLoop
    Socket acceptSocket_;    // 专门用于接受连接的套接字(sock fd)
    Channel acceptChannel_;  // 专门接受连接通道, 监听conn fd
    NewConnectionCallback newConnectionCallback_; // 新建连接回调
    bool listening_;         // 监听状态
    int idleFd_;             // 空闲fd, 用于fd资源不够用时, 可以空一个出来作为新建连接conn fd
};

如果fd资源不够用了，导致accept(2)/accept4(2)创建连接失败，比如达到系统上限，怎么办？
Accetor用了这样一种技术：先申请一个空闲的fd（idleFd_），等到发生由于fd资源不够用时，就把这个备用fd暂时用于accept接收连接，然后再马上关闭，以防止不断产生可读事件（连接请求），从而回调相同的失败代码。及早建立连接后并关闭连接，让程序不会频繁响应同一个连接请求。

Acceptor构造与析构

Acceptor构造时，创建sockfd（套接字），待后续交给TcpServer来start监听套接字。
空闲fd指向文件"/dev/null"，用来解决服务器fd资源耗尽问题。

// Acceptor.cc
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)) // 申请空闲fd
{
    assert(idleFd_ >= 0);
    acceptSocket_.setReuseAddr(true);
    acceptSocket_.setReusePort(reuseport);
    acceptSocket_.bindAddress(listenAddr);
    acceptChannel_.setReadCallback(
            std::bind(&Acceptor::handleRead, this));
}

Acceptor::~Acceptor()
{
    acceptChannel_.disableAll(); // disable all event of the channel
    acceptChannel_.remove(); // remove the channel from poller
    ::close(idleFd_);
}

Acceptor监听

Acceptor包含2类监听：1）监听套接字，即本地ip地址&端口。2）监听通道事件，读事件。

• 为什么不在构造时，就调用listen监听sockfd呢？
  将非必要资源的初始化，延迟到需要时，用户可以通过调用TcpSever::start()来启动。这样，用户可以更灵活控制资源的申请和释放。

// Acceptor.cc
/**
* 监听本地sock fd, 使能监听Channel读事件
*/
void Acceptor::listen()
{
    loop_->assertInLoopThread();
    listening_ = true;
    acceptSocket_.listen();         // 使能监听本地sock fd(ip, port)
    acceptChannel_.enableReading(); // 使能监听通道读事件
}

Acceptor接受连接

Acceptor内部有一个Channel成员，当Poller监听到有Tcp连接请求时，就通过Channel的可读事件，在loop线程，来回调Acceptor::handleRead()。从而将conn fd和IP地址传递给上一层TcpServer，用于创建TcpConnection对象管理Tcp连接。

// Acceptor.cc
/**
* 处理读Channel事件, accept连接
* @note 先accept, 然后将相关资源通过回调交由上一层的TcpServer进行处理(管理)
*/
void Acceptor::handleRead()
{
    loop_->assertInLoopThread();
    InetAddress peerAddr;
    // FIXME loop until no more
    int connfd = acceptSocket_.accept(&peerAddr); // 获取连接fd及对端ip地址
    if (connfd >= 0)
    {
        if (newConnectionCallback_)
        { // 创建新连接回调
            newConnectionCallback_(connfd, peerAddr);
        }
        else
        {
            sockets::close(connfd);
        }
    }
    else
    { // 错误
        LOG_SYSERR << "in Acceptor::handleRead";
        /*
         * Read the section named "The special problem of
         * accept()ing when you can't" in libev's doc.
         * By Marc Lehmann, author of libev.
         *
         * The per-process limit of open file descriptors has been reached.
         */
        if (errno == EMFILE)
        { // 文件描述符资源耗尽错误
            ::close(idleFd_);
            idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL);
            ::close(idleFd_);
            // reopen /dev/null, it dose not matter whether it succeeds or fails.
            idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC);
        }
    }
}

TcpServer类

TcpServer类管理TcpConnection，供用户直接使用，生命周期由用户控制。接口如下，用户只需要设置好callback，然后调用start()即可。

// TcpServer.h
/**
* Tcp Server, 支持单线程和thread-poll模型.
* 接口类, 因此不要暴露太多细节.
*/
class TcpServer : private noncopyable
{
public:
    typedef std::function<void (EventLoop*)> ThreadInitCallback;
    enum Option
    enum Option
    {
        kNoReusePort, // 不允许重用本地端口
        kReusePort,   // 允许重用本地端口
    };

//    TcpServer(EventLoop* loop, const InetAddress& listenAddr);
    TcpServer(EventLoop* loop,
              const InetAddress& listenAddr,
              const std::string& nameArg,
              Option option = kNoReusePort);
    ~TcpServer(); // force out-line dtor, for std::unique_ptr members.

    /**
     * 如果没有监听, 就启动服务器(监听).
     * 多次调用没有副作用.
     * 线程安全.
     */
    void start();

    /**
     * 设置连接回调.
     * 非线程安全.
     */
    void setConnectionCallback(const ConnectionCallback& cb)
    { connectionCallback_ = cb; }

    /**
     * 设置消息回调.
     * 非线程安全.
     */
    void setMessageCallback(const MessageCallback & cb)
    { messageCallback_ = cb; }

    /**
     * 设置写完成回调.
     * 非线程安全.
     */
    void setWriteCompleteCallback(const WriteCompleteCallback& cb)
    { writeCompleteCallback_ = cb; }

注意到并没有连接关闭回调，这是由TcpServer::removeConnection()负责的，进而把工作转交给TcpConnection::connectDestroyed()，用户不可更改设置。

TcpServer的构造与析构

TcpServer构造函数主要工作是为成员申请资源，为各回调设置回调函数

TcpServer::TcpServer(EventLoop* loop,
                     const InetAddress& listenAddr,
                     const std::string& nameArg,
                     Option option)
: loop_(CHECK_NOTNULL(loop)), // 确保loop非空
  ipPort_(listenAddr.toIpPort()),  // 将Ip, port转换为字符串
  name_(nameArg), // 名称
  acceptor_(new Acceptor(loop, listenAddr, option == kReusePort)),
  threadPool_(new EventLoopThreadPool(loop, name_)), // 初始化事件循环线程池
  connectionCallback_(defaultConnectionCallback),    // 连接回调为默认连接回调
  messageCallback_(defaultMessageCallback),          // 消息回调为默认消息回调
  nextConnId_(1)  // 连接id
{
    acceptor_->setNewConnectionCallback(
            std::bind(&TcpServer::newConnection, this, _1, _2)); // 设置新建连接时的回调
}

同样是连接回调，TcpServer::newConnection()和connectionCallback_有何区别？
前者是Acceptor发生连接请求事件时，回调，用来新建一个Tcp连接；后者是在TcpServer内部新建连接即调用TcpServer::newConnection()时，回调connectionCallback_。

TcpServer析构工作内容很简单，主要销毁ConnectionMap中所有Tcp连接，而每个Tcp连接用的是一个TcpConnection对象来管理的。

/**
* 析构TcpServer对象, 销毁ConnectionMap中所有连接
*/
TcpServer::~TcpServer()
{
    loop_->assertInLoopThread();
    LOG_TRACE << "TcpServer::~TcpServer [" << name_ << "] destructing";


    // reset all connection of @c connections_
    for (auto& item : connections_)
    {
        TcpConnectionPtr conn(item.second); // shared_ptr manage TcpConnection
        item.second.reset();
        conn->getLoop()->runInLoop(
                std::bind(&TcpConnection::connectDestroyed, conn));
    }
}

TcpServer启动Tcp服务器，主要完成1）线程池的启动；2）Acceptor监听Tcp连接请求。

线程池需要指定其初始数量，当然，这需要在start()之前调用TcpServer::setThreadNum()设置。

/**
* 启动TcpServer, 初始化线程池, 连接接受器Accept开始监听(Tcp连接请求)
*/
void TcpServer::start()
{
    if (started_.getAndSet(1) == 0)
    {
        threadPool_->start(threadInitCallback_);


        assert(!acceptor_->listening());
        loop_->runInLoop(
                std::bind(&Acceptor::listen, get_pointer(acceptor_)));
    }
}

• TcpServer如何获得新连接的conn fd（accept返回值）？
  TcpServer内部用Acceptor，保存用户提供的Connection-Callback和MessageCallback，新建TcpConnection对象（newConn()）的时候直接传递给TcpConnection的构造函数。

• 如何创建TcpConnection对象？
  新连接请求到达时，Acceptor回调newConnection()，通过TcpServer::newConnection创建一个新TcpConnection对象，用于管理一个Tcp连接。
  即，TcpServer::newConnection回调顺序 EventLoop => Channel => Acceptor => TcpServer

下面是用于创建TcpConnection对象的函数TcpServer::newConnectio()

// TcpServer.cc
/**
* 新建一个TcpConnection对象, 用于连接管理.
* @details 新建的TcpConnection对象会加入内部ConnectionMap.
* @param sockfd accept返回的连接fd (accepted socket fd)
* @param peerAddr 对端ip地址信息
* @note 必须在所属loop线程运行
*/
void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
{
    loop_->assertInLoopThread();
    /* 从EventLoop线程池中，取出一个EventLoop对象构造TcpConnection对象，便于均衡各EventLoop负责的连接数　*/
    EventLoop* ioLoop = threadPool_->getNextLoop(); // next event loop from the event loop thread pool

    /* 设置连接对象名称, 包含基础名称+ip地址+端口号+连接Id
     * 因为要作为ConnectionMap的key, 要确保运行时唯一性 */
    char buf[32];
    snprintf(buf, sizeof(buf), "-%s#%d", ipPort_.c_str(), nextConnId_);
    ++nextConnId_;
    std::string connName = name_ + buf;

    LOG_INFO << "TcpServer::newConnection [" << name_
    << "] - new connection [" << connName
    << "] from " << peerAddr.toIpPort();

    InetAddress localAddr(sockets::getLocalAddr(sockfd)); // 本地ip地址信息
    // FIXME poll with zero timeout to double confirm the new connection
    // FIXME use make_shared if necessary
    /* 新建TcpConnection对象,　并加入ConnectionMap */
    TcpConnectionPtr conn(new TcpConnection(ioLoop, connName, sockfd, localAddr, peerAddr));
    connections_[connName] = conn;
    /* 为新建TcpConnection对象设置各种回调 */
    conn->setConnectionCallback(connectionCallback_); // 连接回调
    conn->setMessageCallback(messageCallback_);       // 消息回调
    conn->setWriteCompleteCallback(writeCompleteCallback_); // 写完成回调
    conn->setCloseCallback( // 关闭连接回调
            std::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe
    /* 确认连接是否已建立, 并初始化连接建立后的状态 */
    ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
}

TcpConnection类

TcpConnection类是muduo最核心的类，唯一默认用shared_ptr来管理的类，唯一继承自enable_shared_from_this的类。这是因为其生命周期模糊：可能在连接断开时，还有其他地方持有它的引用，贸然delete会造成空悬指针。只有确保其他地方没有持有该对象的引用的时候，才能安全地销毁对象。

// TcpConnection.h
/**
* Tcp连接, 为服务器和客户端使用.
* 接口类, 因此不要暴露太多细节.
* 
* @note 继承自std::enable_shared_from_this的类, 可以用getSelf返回用std::shared_ptr管理的this指针
*/
class TcpConnection : noncopyable,
        public std::enable_shared_from_this<TcpConnection> // std::shared_ptr<TcpConnection> getSelf()
{
public:

TcpConnection不提供用户使用的函数，所提供的接口主要给TcpServer/TcpClient使用。TcpConnection的状态有4个：kDisconnected, kConnecting, kConnected, kDisconnecting。TcpConnection使用Channel获得socket上IO事件，它会自己处理writable事件，把readable事件通过MessageCallback传递给客户。TcpConnection拥有TCP socket（Socket类），后者析构函数会close(fd)。

// TcpConnection.h
private:
    enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting }; // TCP连接状态定义
    void handleRead(Timestamp receiveTime); // 处理读事件
    void handleWrite();                     // 处理写事件
    void handleClose();                     // 处理关闭连接事件
    void handleError();                     // 处理错误事件

    /* loop线程中排队发送消息 */
//    void sendInLoop(std::string&& message); // C++11 // add by martin
    void sendInLoop(const StringPiece& message);
    void sendInLoop(const char* message, size_t len);

    /* loop线程中排队关闭写连接 */
    void shutDownInLoop();
//    void shutDownAndForceCloseInLoop(double seconds); // add by martin
    /* loop线程中排队关闭连接 */
    void forceCloseInLoop();

    /*　设置TcpConnection状态 */
    void setState(StateE s) { state_ = s; }

    /* 将状态转换为字符串 */
    const char* stateToString() const;

    /* loop线程中排队开始监听读事件 */
    void startReadInLoop();
    /* loop线程中排队关闭监听读事件 */
    void stopReadInLoop();

    EventLoop* loop_;     // 所属　EventLoop
    std::string name_;    // 名称
    StateE state_; // FIXME: use atomic varaible　//　状态
    bool reading_; // whether the connection is reading // 连接是否正在监听读事件
    // we don't expose those classes to client.
    std::unique_ptr<Socket> socket_;              // 连接套接字, 用于对连接进行底层操作
    std::unique_ptr<Channel> channel_;            // 通道, 用于绑定要监听的事件
    InetAddress localAddr_;                       // 本地IP地址
    InetAddress peerAddr_;                        // 对端IP地址
    ConnectionCallback connectionCallback_;       // 连接回调
    MessageCallback messageCallback_;             // 收到消息回调
    WriteCompleteCallback writeCompleteCallback_; // 写完成回调
    HighWaterMarkCallback highWaterMarkCallback_; // 高水位回调
    CloseCallback closeCallback_;                 // 关闭连接回调
    size_t highWaterMark_;                        // 高水位阈值
    Buffer inputBuffer_;                          // 输入缓冲区
    Buffer outputBuffer_; // FIXME: use list<Buffer> as output buffer.　//　输出缓冲区
    boost::any contex_;                           // 用户自定义参数
    // FIXME: createTime_, lastReceiveTime_
    //        bytesReceived_, bytesSent_
};

TcpConnection表示的是“一次Tcp连接”，不可再生，一旦连接断开，该TcpConnection对象就没用了。TcpConnection没用发起连接的功能，构造函数参数是已经建立好连接的socket fd，初始状态是kConnecting。连接可以是TcpServer或TcpClient发起。

接收到消息时，通过Channel::handleEvent会将可读事件转交给TcpConnection::handleRead处理，而TcpConnection::handleRead又会通过messageCallback_将可读事件转交给TcpServer::messageCallback_，进而传递给用户。

// TcpConnection.cc
/**
* 从输入缓存inputBuffer_读取数据, 交给回调messageCallback_处理
* @param receiveTime 接收到读事件的时间点
* @details 通常是TcpServer/TcpClient运行回调messageCallback_, 将处理机会传递给用户
*/
void TcpConnection::handleRead(Timestamp receiveTime)
{
    loop_->assertInLoopThread();
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno); // 从指定fd读取数据到内部缓冲
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_SYSERR << "TcpConnection::handleRead";
        handleError();
    }
}

断开Tcp连接

断开连接方式

muduo中有2种关闭连接的方式：
1）被动关闭：即对端先关闭连接，本地read(2)返回0，触发关闭逻辑，调用handleClose。
2）主动关闭：利用forceClose()或forceCloseWithDelay()成员函数调用handleClose，强制关闭或强制延时关闭连接。

被动关闭流程见下图，图中“X”表示TcpConnection对象通常在此析构。

Channel与断开连接

Channel中有关关闭连接的事件回调CloseCallback，由Channel::handleEvent()调用，从而触发TcpConnection::handleClose()：

调用链路：
Poller::poll()检测到Channel事件就绪 => EventLoop::loop() =>Channel::handleEvent() => Channel::closeCallback_ => TcpConnection::handleClose()

// Channel.cc
void Channel::handleEvent(Timestamp recevieTime)
{
    std::shared_ptr<void> guard;
    if (tied_)
    {
        guard = tie_.lock();
        if (guard)
        {
            handleEventWithGuard(recevieTime);
        }
    }
    else
    {
        handleEventWithGuard(recevieTime);
    }
}

/**
* 根据不同的激活原因, 调用不同的回调函数
*/
void Channel::handleEventWithGuard(Timestamp receiveTime)
{
    eventHandling_ = true; // 正在处理事件
    LOG_TRACE << reventsToString(); // 打印fd及就绪事件
    if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
    { // fd挂起(套接字已不在连接中), 并且没有数据可读
        if (logHup_)
        { // 打印挂起log
            LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLHUP";
        }
        // 调用关闭回调
        if (closeCallback_) closeCallback_();
    }
    ...
}

TcpConnection与断开连接

包含CloseCallback事件回调，不过是给TcpServer和TcpClient用的，用于通知它们移除所持有的TcpConnectionPtr，而非给普通用户直接用的，普通用户应使用ConnectionCallback。

与断开连接有关的部分：

// TcpConnection.h
public:
    /* 关闭写半连接 */
    void shutdown(); // NOT thread safe, no simultaneous calling
//    void shutdownAndForceCloseAfter(double seconds); // NOT thread safe, no simultaneous calling

    /* 强制关闭连接 */
    void forceClose();
    /* 强制延时关闭连接 */
    void forceCloseWithDelay(double seconds);

    ...

    /* Internal use only */
    void setCloseCallback(const CloseCallback& cb)
    { closeCallback_ = cb; }

    // called when TcpServer accepts a new connection
    void connectEstablished(); // should be called only once per connection
    // called when TcpServer has removed me from its map
    void connectDestroyed(); // should be called only once per connection

private:
    void handleClose();                     // 处理关闭连接事件

• 被动关闭连接
  当收到对端FIN分节时，本地read返回0,，Tcp连接被动关闭，会触发调用本地TcpConnection::handleClose()，其定义如下：

/**
* 处理Tcp连接关闭
* @details 更新状态为kDisconnected, 清除所有事件通道监听
* @note 必须在所属loop线程中运行.
*/
void TcpConnection::handleClose()
{
    loop_->assertInLoopThread(); // 确保在所属loop线程中运行
    LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
    assert(state_ == kConnected || state_ == kDisconnecting);
    // we don't close fd, leave it to dtor, so we can find leaks easily.
    setState(kDisconnected); // 更新Tcp连接状态
    channel_->disableAll();     // 停止监听所有通道事件(读写事件)

    TcpConnectionPtr guardThis(shared_from_this());
    connectionCallback_(guardThis); // 连接回调
    // must be the last line
    closeCallback_(guardThis);      //　关闭连接回调
}

closeCallback_在TcpServer::newConnection()为新连接新建TcpConnection时，已设为TcpServer::removeConnection()，而removeConnection()最终会调用TcpConnection::connectDestroyed()来销毁连接资源。

/**
* 主动销毁当前tcp连接, 移除通道事件
* @note 只有处于已连接状态(kConnected)的tcp连接, 才需要先更新状态, 关闭通道事件监听
*/
void TcpConnection::connectDestroyed()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected) // 只有kConnected的连接, 才有必要采取断开连接动作
    {
        setState(kDisconnected);
        channel_->disableAll(); // 关闭通道事件监听

        connectionCallback_(shared_from_this()); // 调用连接回调
    }
    channel_->remove(); // 从EventLoop和Poller中移除监听通道事件
}

• 主动关闭连接
  按连接方向，有2类关闭方式：1）强制close连接；2）关闭一个方向的连接（读或写方向）。

/**
* 强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
* @details 为防止意外, 动作应该放到loop末尾去做
*/
void TcpConnection::forceClose()
{
    // FIXME: use compare and swap
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        setState(kDisconnecting);
        loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
    }
}

/**
* 在所属loop循环中强制关闭连接, 只对连接为kConnected或kDisconnecting状态才有效
*/
void TcpConnection::forceCloseInLoop()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        // as if we received 0 byte in handleRead()
        handleClose();
    }
}

可以看到，除了状态更新，对于关闭连接的真正操作，被动、主动关闭连接都是由handleClose来完成的。

假设想延迟一段时间再关闭连接，可以调用forceCloseWithDelay()，区别在于交给EventLoop:runAfter()延时运行，而不是交给EventLoop::queueInLoop()。

void TcpConnection::forceCloseWithDelay(double seconds)
{
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        setState(kDisconnecting);
        loop_->runAfter(seconds,
                        makeWeakCallback(shared_from_this(),
                                          &TcpConnection::forceClose)); // not forceCloseInLoop to avoid race condition
    }
}

关闭连接写方向，相当于库函数shutdown(2)

/**
* 关闭连接写方向, 只有已连接状态才有效
*/
void TcpConnection::shutdown()
{
    // FIXME: use compare and swap
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        // FIXME: shared_from_this()?
        loop_->runInLoop(std::bind(&TcpConnection::shutDownInLoop, shared_from_this()));
    }
}

关于close和shutdown区别，参见：https://www.cnblogs.com/fortunely/p/14854299.html

TcpServer与断开连接

当新建一个tcp连接时，TcpServer会调用newConnection创建一个新TcpConnection对象管理Tcp连接，并将对象加入自己的ConnectionMap进行管理。

而当tcp连接断开时，需要调用removeConnection进行移除工作，而removeConnection会将工作转交给removeConnectionInLoop, 确保在所属loop线程中执行。

/**
* 转交给removeConnectionInLoop, 在所属loop线程中执行
*/
void TcpServer::removeConnection(const TcpConnectionPtr &conn)
{
    // FIXME: unsafe
    loop_->runInLoop(std::bind(&TcpServer::removeConnectionInLoop, this, conn));
}

removeConnectionInLoop要做的工作是将要移除的tcp连接对应TcpConnection对象，从ConnectionMap移除，然后销毁该对象。

/**
* 在所属loop线程循环中, 排队移除指定tcp连接 conn
* @param conn 指向待移除tcp连接对应TcpConnection对象
*/
void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn)
{
    loop_->assertInLoopThread();
    LOG_INFO << "TcpServer::removeConnectionInLoop [" << name_
    << "] - connection " << conn->name();
    size_t n = connections_.erase(conn->name()); // 从ConnectionMap中擦除待移除TcpConnection对象
    (void)n;
    assert(n == 1);
    EventLoop* ioLoop = conn->getLoop();
    ioLoop->queueInLoop(
            std::bind(&TcpConnection::connectDestroyed, conn)); // 在所属loop线程中排队销毁TcpConnection对象
}

错误处理

Channel在处理通道事件handleEvent时，如果发生错误（既不是读取数据，也不是写数据完成），

调用链路：
Channel::handleEvent() => 检测到错误，调用Channel::errorCallback_ => TcpConnection::handleError()

/**
* 处理tcp连接错误, 打印错误log
* @details 从tcp协议栈获取错误信息
*/
void TcpConnection::handleError()
{
    int err = sockets::getSocketError(channel_->fd());
    LOG_ERROR << "TcpConnection::handleError [" << name_
    << "] - SO_ERROR = " << err << " " << strerror_tl(err);
}

发送数据

用于发送数据的TcpConnection::send()重载了下面几个版本

public:
    ...
    /* 发送消息给连接对端 */
//    void send(std::string&& mesage); // C++11
    void send(const void* message, int len);
    void send(const StringPiece& message);
//    void send(Buffer&& message);
    void send(Buffer* message); // this one will swap data

3个send() 重载版本，最终都会转交给sendInLoop(const char*, int)，在所属loop线程中执行发送工作。


// 转发给send(const StringPiece&), 最终转交给sendInLoop(const char*, int)
void TcpConnection::send(const void *message, int len)
{
    send(StringPiece(static_cast<const char*>(message), len));
}

/**
* 转交给 sendInLoop(const char*, int)
* 发送消息给对端, 允许在其他线程调用
* @param message 要发送的消息. StringPiece兼容C/C++风格字符串, 二进制缓存, 提供统一字符串接口
*/
void TcpConnection::send(const StringPiece& message)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        { // 当前线程是所属loop线程
            sendInLoop(message);
        }
        else
        { // 当前线程并非所属loop线程
            void (TcpConnection::*fp)(const StringPiece& message);
            fp = &TcpConnection::sendInLoop;
            loop_->runInLoop(
                    std::bind(fp,
                              this, // FIXME
                              message.as_string()));
        }
    }
}

// 转交给sendInLoop(const char*, int)
void TcpConnection::sendInLoop(const StringPiece &message)
{
    sendInLoop(message.data(), message.size());
}

// 转交给sendInLoop(const char*, int)
// FIXME efficiency!!!
void TcpConnection::send(Buffer *buf)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        {
            // send all readable bytes
            sendInLoop(buf->peek(), buf->readableBytes());
            buf->retrieveAll();
        }
    }
}

sendInLoop定义如下，主要是向对端发送一次数据，如果发送完一次，就进行一次回调；如果待发送数据超高水位，就进行高水位回调；如果发生错误，就进行错误回调。

/**
* 在所属loop线程中, 发送data[len]
* @param data 要发送的缓冲区首地址
* @param len　要发送的缓冲区大小(bytes)
* @details 发生write错误, 如果发送缓冲区未满,　对端已发FIN/RST分节 表明tcp连接发生致命错误(faultError为true)
*/
void TcpConnection::sendInLoop(const char *data, size_t len)
{
    loop_->assertInLoopThread();
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    if (state_ == kDisconnected) // 如果已经断开连接(kDisconnected), 就无需发送, 打印log(LOG_WARN)
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }

    // write一次, 往对端发送数据, 后面再看是否发生错误, 是否需要高水位回调
    // if no thing output queue, try writing directly
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    { // 如果通道没有使能监听写事件, 并且outputBuffer　没有待发送数据, 就直接通过socket写
        nwrote = sockets::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeCompleteCallback_)
            {
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
        }
        else // nwrote < 0, error
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK) // EWOULDBLOCK: 发送缓冲区已满, 且fd已设为nonblocking
            { // O_NONBLOCK fd, write block but return EWOULDBLOCK error
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE || errno == ECONNRESET) // FIXME: any others?
                { // EPIPE: reading end is closed; ECONNRESET: connection reset by peer
                    faultError = true;
                }
            }
        }
    }

    // 处理剩余待发送数据
    assert(remaining <= len);
    if (!faultError && remaining > 0) // 没有故障, 并且还有待发送数据, 可能是发送太快, 对方来不及接收
    { // no error and data remaining to be written
        size_t oldLen = outputBuffer_.readableBytes(); // Buffer中待发送数据量

        if (oldLen + remaining >= highWaterMark_ // Buffer及当前要发送的数据量之和 超 高水位(highWaterMark)
        && oldLen < highWaterMark_ // 单独的Buffer中待发送数据量 未超 高水位
        && highWaterMarkCallback_)
        {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
        }
        // append data to be written to the output buffer
        outputBuffer_.append(static_cast<const char*>(data) + nwrote, remaining);
        // enable write event for channel_
        if (!channel_->isWriting())
        { // 如果没有在监听通道写事件, 就使能通道写事件
            channel_->enableWriting();
        }
    }
}

用户使用TcpServer与客户端进行通信

假设客户端请求与服务器建立连接，用户（库的使用者）如何在建立连接、发送数据时，执行自定义任务？
可以在TcpServer启动前，利用TcpServer::setConnectionCallback()、TcpServer::setMessageCallback()等设置回调接口注册任务，等Tcp连接达到指定状态后，会回调用户任务。

示例：使用TcpServer接口类创建Tcp连接
例如，用户自定义服务EchoServer回传收到的客户端消息，EchoServer使用TcpServer注册连接回调（setConnectionCallback）和消息回调（setMessageCallback）。
如果需要分阶段向客户端发送数据，还可以注册写完成回调（setWriteCompleteCallback）。

// examples/simple/echo
EchoServer::EchoServer(muduo::net::EventLoop *loop, const muduo::net::InetAddress &listenAddr)
: server_(loop, listenAddr, "EchoServer")
{
    server_.setConnectionCallback(
            std::bind(&EchoServer::onConnection, this, _1));
    server_.setMessageCallback(
            std::bind(&EchoServer::onMessage, this, _1, _2, _3));

    LOG_INFO << server_.name() << " listen on " << listenAddr.toIpPort();
}

void EchoServer::start()
{
    server_.start();
}

void EchoServer::onConnection(const muduo::net::TcpConnectionPtr &conn)
{
    LOG_INFO << "EchoServer - " << conn->peerAddress().toIpPort() << " -> "
    << conn->localAddress().toIpPort() << " is "
    << (conn->connected() ? "UP" : "DOWN");
}

void EchoServer::onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp time)
{
    muduo::string msg(buf->retrieveAllAsString());
    LOG_INFO << conn->name() << " echo " << msg.size() << " bytes, "
    << "data received at " << time.toString();
    conn->send(msg);
}

int main()
{
    LOG_INFO << "pid = " << getpid();
    muduo::net::EventLoop loop; // 创建EventLoop对象
    muduo::net::InetAddress listenAddr(2007); // 创建包含ip地址、端口号的对象
    EchoServer server(&loop, listenAddr); // 创建用户自定义EchoServer对象
    server.start();  // 启动EchoServer服务器
    loop.loop();     // 启动loop循环
    return 0;
}

重要知识点

宏函数CHECK_NOTNULL

这是一个模板函数重定义以后的宏函数，可以有效检查指针是否为NULL。如果是就直接终止程序（LOG_FATAL），并记录log，避免后续意外情况发生。

// Logging.h

//    Taken from glog/logging.h
//
//    Check that the input is non NULL. This very useful in constructor
//    initializer lists

#define CHECK_NOTNULL(val) 
  ::muduo::CheckNotNull(__FILE__, __LINE__, "'" #val "' Must be non NULL", (val))

// A small helper for CHECK_NOTNULL()
template <typename T>
T* CheckNotNull(Logger::SourceFile file, int line, const char* names, T* ptr)
{
    if (ptr == NULL)
    {
        Logger(file, line, Logger::FATAL).stream() << names;
    }
    return ptr;
}

函数模板get_pointer

用shared_ptr, unique_ptr自带的get()方法，从中取得raw pointer（原生指针）。当然，可以直接调用shared_ptr, unique_ptr，用函数模板重载可以使其含义明确，而且同一的接口方便参数类型可以在允许的类型中切换。

/* get raw pointer from std::shared_ptr */
template<typename T>
inline T* get_pointer(const std::shared_ptr<T>& ptr)
{
    return ptr.get();
}

/* get raw pointer from std::unique_ptr */
template<typename T>
inline T* get_pointer(const std::unique_ptr<T>& ptr)
{
    return ptr.get();
}

客户端：
可以像下面这样，安全的异步调用acceptor_->listen()，注意acceptor_是std::unique_ptr，不能直接拷贝，传递给runInLoop的实参是raw pointer

loop_->runInLoop(std::bind(&Acceptor::listen, get_pointer(acceptor_)));

类模板WeakCallback<>

当我们有一个shared_ptr指针object指向一个对象，那么如何安全地调用某个成员函数func呢？
比如，这个shared_ptr object来自shared_from_this()，如果用object->func()，只能同步调用，没办法异步调用（回调），因为等到异步调用的时候，原来的object对象可能已经释放。此时，可以考虑将object转化为weak_ptr，保存weak_ptr，在需要先提升为shared_ptr，再调用之。

函数子类模板WeakCallback就是这样一种思想。用std::weak_ptr保存一个对象object，std::function保存一个不定参数的函数对象function，并重载函数调用运算符operator()，用于安全地将object作为function的参数来调用。

/**
* 以一种安全方式调用 形如 function(object, ...), 如果object指向一个class, 那么相当于即object->function()
*/
template<typename CLASS, typename... ARGS> // ARGS是可变参数模板参数包
class WeakCallback
{
public:
    WeakCallback(const std::weak_ptr<CLASS>& object, //　weak_ptr安全地使用传入实參
                 const std::function<void (CLASS*, ARGS...)>& function) // function参数个数不确定
                 : object_(object), function_(function)
    {
    }

    // Default dtor, copy ctor and assignment are okay

    void operator()(ARGS&&... args) const // args参数包
    {
        // 安全地使用weak_ptr所指对象
        std::shared_ptr<CLASS> ptr(object_.lock()); // 先lock, 再提升为shared_ptr, 只在local作用域影响引用计数
        if (ptr)
        {
            function_(ptr.get(), std::forward<ARGS>(args)...);
        }
//        else
//        {
//            LOG_TRACE << "expired";
//        }
    }

private:
    std::weak_ptr<CLASS> object_;
    std::function<void (CLASS*, ARGS...)> function_;
};

为了简化使用，可以通过重载函数makeWeakCallback的2个版本，制造一个临时WeakCallback对象，用来保存要调用的函数及参数。

/* 制造一个临时WeakCallback对象, 便于调用object->function(...) */
template<typename CLASS, typename... ARGS>
WeakCallback<CLASS, ARGS...> makeWeakCallback(const std::shared_ptr<CLASS>& object,
                                              void (CLASS::*function)(ARGS...)) // 非const 成员函数
{
    return WeakCallback<CLASS, ARGS...>(object, function);
}

/* 制造一个临时WeakCallback对象, 便于调用object->function(...) const */
template<typename CLASS, typename ... ARGS>
WeakCallback<CLASS, ARGS...> makeWeakCallback(const std::shared_ptr<CLASS>& object,
                                              void (CLASS::*function)(ARGS...) const) // const 成员函数
{
    return WeakCallback<CLASS, ARGS...>(object, function);
}

客户端：可以像下面那样，安全地异步调用shared_from_this()->forceCloseInLoop()。

loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));