muduo库TcpConnection模块详解—

muduo库TcpConnection模块详解——C++

news2025/5/19 15:38:59

muduo库中的`TcpConnection`模块详解

TcpConnection是muduo库中处理TCP连接的核心模块，负责管理单个TCP连接的生命周期、数据读写、状态转换以及事件回调。每个TCP连接对应一个TcpConnection对象，其设计体现了高性能、线程安全和灵活回调的特点。

一、核心职责

连接生命周期管理
管理TCP连接的建立、活跃状态、关闭过程及销毁。
数据读写与缓冲
通过inputBuffer和outputBuffer处理数据的接收与发送，应对粘包/半包问题。
事件回调机制
提供用户可配置的回调函数（如消息到达、连接关闭、发送完成等）。
线程安全保证
确保所有操作在绑定的EventLoop线程中执行，避免竞态条件。

二、关键数据结构与成员

1. 类定义概览

class TcpConnection : public std::enable_shared_from_this<TcpConnection> {
public:
    // 连接状态枚举
    enum State { kConnecting, kConnected, kDisconnecting, kDisconnected };

    // 回调函数类型定义
    using MessageCallback = std::function<void(const TcpConnectionPtr&, Buffer*, Timestamp)>;
    
    using CloseCallback = std::function<void(const TcpConnectionPtr&)>;

private:
    EventLoop* loop_;            // 所属EventLoop（即IO线程）
    State state_;                // 当前连接状态
    std::unique_ptr<Socket> socket_; // 管理的Socket对象
    std::unique_ptr<Channel> channel_; // 关联的Channel，用于事件监听
    Buffer inputBuffer_;         // 接收数据缓冲区
    Buffer outputBuffer_;        // 发送数据缓冲区
    MessageCallback messageCallback_; // 消息到达回调
    CloseCallback closeCallback_;    // 连接关闭回调
};

2. 成员说明

loop_
指向绑定的EventLoop，所有操作必须在该线程执行。
state_
管理连接状态，确保状态转换的合法性（如不允许在已关闭的连接上发送数据）。
socket_和channel_
封装底层socket文件描述符及其事件监听，channel_注册到loop_中处理读写事件。
inputBuffer_和outputBuffer_
应用层缓冲区，用于暂存接收和待发送的数据。
回调函数
用户通过setMessageCallback()等接口设置自定义逻辑。

三、生命周期管理

1. 连接建立

创建时机
由TcpServer或TcpClient在连接建立时创建TcpConnection对象。
初始化流程
- 设置socket为非阻塞模式。
- 绑定Channel到EventLoop，注册可读事件。
- 状态设为kConnecting，随后通过connectEstablished()转为kConnected。

2. 连接关闭

主动关闭
用户调用shutdown()，发送剩余数据后关闭写端（SHUT_WR）。
被动关闭
收到FIN包（读返回0）时，触发handleClose()。
状态转换
kConnected → kDisconnecting → kDisconnected，最终调用closeCallback_通知上层。

3. 对象销毁

使用shared_ptr管理生命周期，通过shared_from_this()确保回调中对象存活。
连接关闭后，TcpServer或TcpClient从连接列表中移除该对象的引用。

四、数据读写流程

1. 数据接收

可读事件处理

void TcpConnection::handleRead(Timestamp receiveTime) {
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0) {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    } else if (n == 0) { // 对端关闭连接
        handleClose();
    } else { // 错误处理
        handleError();
    }
}

用户处理逻辑
通过messageCallback_将数据传递给用户，用户从inputBuffer_解析完整消息。

2. 数据发送

发送入口

void TcpConnection::send(const std::string& message) {
    if (state_ == kConnected) {
        if (loop_->isInLoopThread()) {
            sendInLoop(message.data(), message.size());
        } else {
            loop_->runInLoop([this, msg = message] { 
                sendInLoop(msg.data(), msg.size()); 
            });
        }
    }
}

实际发送逻辑

void TcpConnection::sendInLoop(const void* data, size_t len) {
    if (channel_->isWriting() && outputBuffer_.readableBytes() == 0) {
        // 直接尝试写入socket
        ssize_t n = ::write(channel_->fd(), data, len);
        if (n < 0) { /* 错误处理 */ }
        if (static_cast<size_t>(n) < len) {
            // 剩余数据加入outputBuffer_并注册可写事件
            outputBuffer_.append(static_cast<const char*>(data)+n, len-n);
            channel_->enableWriting();
        }
    } else {
        // 数据直接追加到outputBuffer_
        outputBuffer_.append(data, len);
        if (!channel_->isWriting()) {
            channel_->enableWriting();
        }
    }
}

可写事件处理

void TcpConnection::handleWrite() {
    ssize_t n = ::write(channel_->fd(), outputBuffer_.peek(), outputBuffer_.readableBytes());
    if (n > 0) {
        outputBuffer_.retrieve(n);
        if (outputBuffer_.readableBytes() == 0) {
            channel_->disableWriting(); // 避免忙等待
            if (state_ == kDisconnecting) {
                shutdownInLoop(); // 关闭写端
            }
        }
    }
}

五、状态管理与异常处理

1. 状态转换图

kConnecting → kConnected → kDisconnecting → kDisconnected
       ↑_________________________|

关键转换点：
- connectEstablished()：kConnecting → kConnected
- shutdown()：kConnected → kDisconnecting
- handleClose()：kDisconnecting → kDisconnected

2. 异常处理

socket错误
handleError()调用closeCallback_并关闭连接。
对方意外断开
readFd()返回0时触发handleClose()。
发送失败处理
根据errno判断是否为可恢复错误（如EAGAIN），否则关闭连接。

六、线程安全机制

跨线程调用保护
所有非IO线程的操作通过loop_->runInLoop()转移到IO线程执行。
智能指针管理
使用shared_from_this()确保回调执行期间对象不被销毁。
状态变更原子性
状态state_的修改仅在IO线程中进行，无需加锁。

七、高级功能

TCP_NODELAY选项
通过setTcpNoDelay()禁用Nagle算法，减少小数据包延迟。
Keep-Alive机制
setKeepAlive()启用TCP保活探测，检测死连接。
水位线控制
设置高水位回调（highWaterMarkCallback_），防止发送缓冲区堆积。

八、典型使用示例

// 创建TcpServer并设置回调
TcpServer server(&loop, InetAddress(9876), "EchoServer");
server.setConnectionCallback([](const TcpConnectionPtr& conn) {
    if (conn->connected()) {
        conn->setTcpNoDelay(true); // 启用TCP_NODELAY
    }
});
server.setMessageCallback([](const TcpConnectionPtr& conn, Buffer* buf, Timestamp) {
    conn->send(buf->retrieveAllAsString()); // 回显数据
});
server.start();
loop.loop();