AJohn Blog

外观

9. TCPconnection

约 5588 字大约 19 分钟

2026-04-19

TcpConnection 简介

TcpConnection 类可以理解为 muduo 中对一条 TCP 连接的完整封装。

如果说：

  • Acceptor 负责“接进来一个新连接”
  • Channel 负责“监听某个 fd 上发生了什么事件”
  • EventLoop / Poller 负责“把事件分发出去”

那么 TcpConnection 就是把这些东西串起来的 连接实体。它负责一条连接从建立、读写、半关闭、关闭到销毁的整个生命周期。

它内部不仅保存了：

  • 连接对应的 Socket
  • 连接对应的 Channel
  • 连接的输入输出缓冲区

还保存了用户注册的各种回调函数，比如：

  • 连接建立回调
  • 消息到达回调
  • 写完成回调
  • 高水位回调
  • 关闭回调

可以把它理解为：

  • Socket：真正和内核通信的 fd 封装
  • Channel：负责监听这个 fd 的事件
  • Buffer：负责缓存收发的数据
  • TcpConnection：负责管理这条连接的完整生命周期
thread(main)
 └── TcpServer
      ├── mainLoop
      │    ├── Poller
      │    │    ├── wakeupChannel
      │    │    └── acceptChannel
      │    └── Acceptor
      │         └── Socket(listenfd)
      │
      ├── EventLoopThreadPool
      │    ├── subThread1 -> subLoop1 -> Poller -> Channel(connfd...)
      │    ├── subThread2 -> subLoop2 -> Poller -> Channel(connfd...)
      │    └── subThread3 -> subLoop3 -> Poller -> Channel(connfd...)
      │
      └── connections_
           ├── TcpConnection1
           ├── TcpConnection2
           └── TcpConnection3

TcpConnection 类重要的成员变量

1. 重要成员

  • EventLoop *loop_：这个 TcpConnection 所属的事件循环。
    在多 Reactor 模型下，它通常指向 subLoop；在单 Reactor 模型下，可能指向 baseLoop。
    这个 loop_ 决定了该连接的事件和回调应该在哪个线程中执行。

  • const std::string name_：连接的名字。
    通常由服务器为每条连接生成唯一标识，便于日志打印和调试。

  • std::atomic_int state_：连接当前的状态。
    它是一个原子状态机，用来表示连接处于：

    • kDisconnected
    • kConnecting
    • kConnected
    • kDisconnecting

  • bool reading_：表示当前连接是否正在监听读事件。
    这个成员在当前代码里主要用于标记读状态。

  • std::unique_ptr<Socket> socket_：连接对应的 socket 封装。
    负责执行底层的 shutdownWrite()、setKeepAlive() 等操作。

  • std::unique_ptr<Channel> channel_：连接对应的 Channel。
    负责把这个连接的 fd 注册到 Poller，并监听读写关闭错误等事件。

  • const InetAddress localAddr_：本地地址。
    也就是这个连接所在 socket 的本机 IP 和端口。

  • const InetAddress peerAddr_：对端地址。
    也就是客户端的 IP 和端口。

  • ConnectionCallback connectionCallback_：连接建立/状态变化时的回调。

  • MessageCallback messageCallback_：有消息到达时的回调。

  • WriteCompleteCallback writeCompleteCallback_：消息全部发送完成后的回调。

  • HighWaterMarkCallback highWaterMarkCallback_：输出缓冲区达到高水位时的回调。

  • CloseCallback closeCallback_：连接关闭时的回调。

  • size_t highWaterMark_：高水位阈值。
    当输出缓冲区积压的数据超过这个值时，会触发高水位回调，提醒上层发送压力太大。

  • Buffer inputBuffer_：输入缓冲区。
    用于保存从对端读到的数据。

  • Buffer outputBuffer_：输出缓冲区。
    用于保存暂时还没发完的数据。

TcpConnection.h
EventLoop *loop_; // 这里是baseloop还是subloop由TcpServer中创建的线程数决定 若为多Reactor 该loop_指向subloop 若为单Reactor 该loop_指向baseloop
const std::string name_;
std::atomic_int state_;
bool reading_;//连接是否在监听读事件

// Socket Channel 这里和Acceptor类似    Acceptor => mainloop    TcpConnection => subloop
std::unique_ptr<Socket> socket_;
std::unique_ptr<Channel> channel_;

const InetAddress localAddr_;
const InetAddress peerAddr_;

// 这些回调TcpServer也有 用户通过写入TcpServer注册 TcpServer再将注册的回调传递给TcpConnection TcpConnection再将回调注册到Channel中
ConnectionCallback connectionCallback_;       // 有新连接时的回调
MessageCallback messageCallback_;             // 有读写消息时的回调
WriteCompleteCallback writeCompleteCallback_; // 消息发送完成以后的回调
HighWaterMarkCallback highWaterMarkCallback_; // 高水位回调
CloseCallback closeCallback_; // 关闭连接的回调
size_t highWaterMark_; // 高水位阈值

// 数据缓冲区
Buffer inputBuffer_;    // 接收数据的缓冲区
Buffer outputBuffer_;   // 发送数据的缓冲区 用户send向outputBuffer_发

连接状态（StateE）

TcpConnection.h
enum StateE
{
    kDisconnected, // 已经断开连接
    kConnecting,   // 正在连接
    kConnected,    // 已连接
    kDisconnecting // 正在断开连接
};

状态含义

  • kConnecting：连接刚创建，还处于建立阶段。
  • kConnected：连接已经建立，可以正常收发数据。
  • kDisconnecting：上层已经调用了 shutdown()，正在等待发送缓冲区的数据发送完毕后再优雅关闭。
  • kDisconnected：连接已经断开，不能再发送数据。

为什么需要状态机？

因为 TCP 连接不是“连上/断开”这么简单。 比如：

  • 对方关闭了连接
  • 本地想优雅关闭写端
  • 发送缓冲区还有未发完的数据
  • 连接已经销毁，不能再访问

状态机可以保证这些流程不会乱掉。

TcpConnection 类重要的成员方法

1. 构造 / 析构

TcpConnection.h
TcpConnection(EventLoop *loop,
              const std::string &nameArg,
              int sockfd,
              const InetAddress &localAddr,
              const InetAddress &peerAddr);
~TcpConnection();

构造函数主要做了几件事：

  1. 检查 loop 是否为空；
  2. 保存连接名字；
  3. 初始化连接状态为 kConnecting；
  4. 创建 Socket 和 Channel；
  5. 保存本地和对端地址；
  6. 设置默认高水位阈值；
  7. 给 Channel 注册读、写、关闭、错误回调；
  8. 打开 SO_KEEPALIVE。

它把“一个 fd + 一个事件循环 + 一组缓冲区 + 一组回调”组合成了一条完整连接。

2. getLoop() / name() / 地址获取 / connected()

TcpConnection.h
EventLoop *getLoop() const { return loop_; }
const std::string &name() const { return name_; }
const InetAddress &localAddress() const { return localAddr_; }
const InetAddress &peerAddress() const { return peerAddr_; }

bool connected() const { return state_ == kConnected; }

这些函数主要是查询连接的基本信息。

  • getLoop()：拿到连接所属的 EventLoop。
  • name()：拿到连接名称。
  • localAddress() / peerAddress()：拿到本地和对端地址。
  • connected()：判断当前连接是否处于已连接状态。

3. send(const std::string &buf)

TcpConnection.h
void send(const std::string &buf);

这个函数用于发送数据。

如果当前就在连接所属的事件循环线程中，就直接发送；

如果不在，就把发送任务投递到所属线程中执行。

提示

为什么要这样？

因为 TcpConnection 和它的 Channel 必须在正确的 EventLoop 线程中操作，不能随便跨线程直接改状态。

4. sendInLoop(const void *data, size_t len)

TcpConnection.h
void sendInLoop(const void *data, size_t len);

这个函数是真正执行发送逻辑的地方。

  1. 先尝试直接写到 socket；
  2. 如果没写完，把剩余数据放到 outputBuffer_；
  3. 需要时注册写事件 EPOLLOUT，等待后续继续发送；
  4. 如果全部写完，触发写完成回调；
  5. 如果输出缓冲区积压过多，触发高水位回调。

提示

为什么要有输出缓冲区？

因为应用层发送速度通常比内核发送速度快。 如果一次写不完，就必须把未发送完的数据缓存起来，等 socket 可写时再继续发。

5. sendFile(int fileDescriptor, off_t offset, size_t count)

TcpConnection.h
void sendFile(int fileDescriptor, off_t offset, size_t count);

这个函数用于零拷贝发送文件数据。

如果连接处于已连接状态，就把 sendfile() 相关任务放到正确的事件循环线程中执行。

提示

为什么有这个接口？

它可以更高效地把文件内容直接发送到 socket，减少用户态拷贝。

6. sendFileInLoop(int fileDescriptor, off_t offset, size_t count)

TcpConnection.h
void sendFileInLoop(int fileDescriptor, off_t offset, size_t count);

这个函数是真正执行 sendfile() 的地方。

它的工作方式

  1. 先检查连接状态；
  2. 如果当前没有待发送缓冲数据，就尝试直接用 sendfile()；
  3. 如果没有一次发完，就把剩余部分继续投递到事件循环中；
  4. 如果全部发送完，触发写完成回调。

7. shutdown()

TcpConnection.h
void shutdown();

这个函数用于关闭写方向，也就是 TCP 的半关闭。

如果当前处于已连接状态，就把状态切成 kDisconnecting，然后在所属线程中执行 shutdownInLoop()。

提示

为什么不是直接 close？

因为这里想要的是 优雅关闭： 先把输出缓冲区剩余数据发送完，再关闭写端。

8. shutdownInLoop()

TcpConnection.h
void shutdownInLoop();

这个函数真正执行写端关闭。

如果当前 Channel 已经不再监听写事件，说明输出缓冲区的数据已经发完了， 这时候就可以调用 socket_->shutdownWrite()，关闭写方向。

9. connectEstablished()

TcpConnection.h
void connectEstablished();

这个函数表示连接已经建立完成。

  1. 把状态改成 kConnected；
  2. 通过 tie() 绑定自身生命周期，防止回调执行时对象被销毁；
  3. 注册读事件；
  4. 触发连接建立回调。

连接建立后，Channel 才真正开始监听这个 socket 的读事件，后续数据才能被收到。

10. connectDestroyed()

TcpConnection.h
void connectDestroyed();

这个函数表示连接即将销毁。

  1. 如果连接还处于 kConnected，先改成 kDisconnected；
  2. 取消所有事件监听；
  3. 触发连接回调，通知上层连接状态变化；
  4. 最后把 Channel 从 Poller 中移除。

提示

为什么最后要 remove()？

因为连接彻底结束后，Channel 就不应该再被 Poller 管理了。

11. handleRead(Timestamp receiveTime)

TcpConnection.h
void handleRead(Timestamp receiveTime);

这个函数是读事件的回调。

  1. 从 socket 读数据到输入缓冲区；
  2. 如果读到了数据，就触发消息回调；
  3. 如果读到 0，说明对端关闭了连接；
  4. 如果出错，就处理错误。

提示

为什么要先读进 inputBuffer_？

因为 TCP 是字节流，应用层收到的往往不是一个完整“消息”，而是连续的数据流。 先放进缓冲区，再由上层按协议解析更安全。

12. handleWrite()

TcpConnection.h
void handleWrite();

这个函数是写事件的回调。

  1. 检查当前是否真的在监听写事件；
  2. 把输出缓冲区的数据写到 socket；
  3. 如果写完了，就取消写事件监听；
  4. 触发写完成回调；
  5. 如果之前进入了 kDisconnecting，现在可以真正关闭写端了。

它负责把 outputBuffer_ 里的数据尽量写出去，并在写完后恢复正常状态。

13. handleClose()

TcpConnection.h
void handleClose();

这个函数用于处理连接关闭事件。

  1. 打日志；
  2. 状态改成 kDisconnected；
  3. 取消所有事件监听；
  4. 触发连接回调；
  5. 调用关闭回调，把销毁请求交给上层。

提示

为什么 closeCallback_ 必须放在最后？

因为它通常会触发连接移除和销毁逻辑。 一旦执行了关闭回调，这个对象后续就可能被销毁，所以它必须放在最后一行。

14. handleError()

TcpConnection.h
void handleError();

这个函数用于处理 socket 错误。

通过 getsockopt(SO_ERROR) 查询 socket 的具体错误原因，然后打印日志。

TcpConnection 和 Channel、Socket、Buffer 的关系

可以这样理解：

  • Socket：负责底层系统调用，比如关闭写端、设置 keepalive
  • Channel：负责监听 fd 的读写关闭错误事件
  • Buffer：负责缓存输入输出数据
  • TcpConnection：把这三者拼成一条完整连接，并管理连接生命周期

TcpConnection 的整体工作流程

1. 创建连接

TcpServer::newConnection() 创建 TcpConnection，初始化其 Socket、Channel 和缓冲区。

2. 建立连接

调用 connectEstablished()：

  • 把状态设为 kConnected
  • 绑定 tie
  • 开始监听读事件
  • 执行连接建立回调

3. 收消息

当 socket 可读时：

  • Channel 触发 handleRead()
  • TcpConnection 把数据读到 inputBuffer_
  • 执行 messageCallback_

4. 发消息

上层调用 send()：

  • 如果能直接写，就直接写
  • 写不完就放进 outputBuffer_
  • 注册写事件，等待后续继续发送

5. 关闭连接

上层调用 shutdown() 或对端关闭连接：

  • 状态机转移
  • 停止监听事件
  • 执行关闭流程
  • 最终调用 connectDestroyed()

TcpConnection 就像一条“完整的会话管道”：

  • 它知道这条连接是谁和谁之间的
  • 它知道什么时候读、什么时候写
  • 它知道数据先放哪儿
  • 它知道什么时候关闭
  • 它还知道该通知谁

总结

TcpConnection 的核心职责

  • 封装一条 TCP 连接的生命周期
  • 管理对应的 Socket 和 Channel
  • 维护输入输出缓冲区
  • 处理读、写、关闭、错误事件
  • 提供连接、消息、写完成、高水位、关闭等回调
  • 保证所有操作在正确的 EventLoop 线程中执行

TcpConnection 是 muduo 中对一条 TCP 连接的完整封装，它把 Socket、Channel、Buffer 和各种回调组合起来，负责连接从建立到销毁的整个生命周期。

TcpConnection 源码

TcpConnection.h
#pragma once

#include <memory>
#include <string>
#include <atomic>

#include "noncopyable.h"
#include "InetAddress.h"
#include "Callbacks.h"
#include "Buffer.h"
#include "Timestamp.h"

class Channel;
class EventLoop;
class Socket;

/**
 * TcpServer => Acceptor => 有一个新用户连接，通过accept函数拿到connfd
 * => TcpConnection设置回调 => 设置到Channel => Poller => Channel回调
 **/

class TcpConnection : noncopyable, public std::enable_shared_from_this<TcpConnection>
{
public:
    TcpConnection(EventLoop *loop,
                  const std::string &nameArg,
                  int sockfd,
                  const InetAddress &localAddr,
                  const InetAddress &peerAddr);
    ~TcpConnection();

    EventLoop *getLoop() const { return loop_; }
    const std::string &name() const { return name_; }
    const InetAddress &localAddress() const { return localAddr_; }
    const InetAddress &peerAddress() const { return peerAddr_; }

    bool connected() const { return state_ == kConnected; }

    // 发送数据
    void send(const std::string &buf);
    void sendFile(int fileDescriptor, off_t offset, size_t count); 
    
    // 关闭半连接
    void shutdown();

    void setConnectionCallback(const ConnectionCallback &cb)
    { connectionCallback_ = cb; }
    void setMessageCallback(const MessageCallback &cb)
    { messageCallback_ = cb; }
    void setWriteCompleteCallback(const WriteCompleteCallback &cb)
    { writeCompleteCallback_ = cb; }
    void setCloseCallback(const CloseCallback &cb)
    { closeCallback_ = cb; }
    void setHighWaterMarkCallback(const HighWaterMarkCallback &cb, size_t highWaterMark)
    { highWaterMarkCallback_ = cb; highWaterMark_ = highWaterMark; }

    // 连接建立
    void connectEstablished();
    // 连接销毁
    void connectDestroyed();

private:
    enum StateE
    {
        kDisconnected, // 已经断开连接
        kConnecting,   // 正在连接
        kConnected,    // 已连接
        kDisconnecting // 正在断开连接
    };
    void setState(StateE state) { state_ = state; }

    void handleRead(Timestamp receiveTime);
    void handleWrite();//处理写事件
    void handleClose();
    void handleError();

    void sendInLoop(const void *data, size_t len);
    void shutdownInLoop();
    void sendFileInLoop(int fileDescriptor, off_t offset, size_t count);
    EventLoop *loop_; // 这里是baseloop还是subloop由TcpServer中创建的线程数决定 若为多Reactor 该loop_指向subloop 若为单Reactor 该loop_指向baseloop
    const std::string name_;
    std::atomic_int state_;
    bool reading_;//连接是否在监听读事件

    // Socket Channel 这里和Acceptor类似    Acceptor => mainloop    TcpConnection => subloop
    std::unique_ptr<Socket> socket_;
    std::unique_ptr<Channel> channel_;

    const InetAddress localAddr_;
    const InetAddress peerAddr_;

    // 这些回调TcpServer也有 用户通过写入TcpServer注册 TcpServer再将注册的回调传递给TcpConnection TcpConnection再将回调注册到Channel中
    ConnectionCallback connectionCallback_;       // 有新连接时的回调
    MessageCallback messageCallback_;             // 有读写消息时的回调
    WriteCompleteCallback writeCompleteCallback_; // 消息发送完成以后的回调
    HighWaterMarkCallback highWaterMarkCallback_; // 高水位回调
    CloseCallback closeCallback_; // 关闭连接的回调
    size_t highWaterMark_; // 高水位阈值

    // 数据缓冲区
    Buffer inputBuffer_;    // 接收数据的缓冲区
    Buffer outputBuffer_;   // 发送数据的缓冲区 用户send向outputBuffer_发
};