企业管理网站百度指数搜索热度

---

前言

今天总结TcpConnection类的读写事件。

一、TcpConnection的读

当Poller检测到套接字的Channel处于可读状态时，会调用Channel的回调函数，回调函数中根据不同激活原因调用不同的函数，这些函数都由TcpConnection在创建Channel之初提供，当可读时，调用TcpConnection的可读函数handleRead，而在这个函数中，读缓冲区就会从内核的tcp缓冲区读取数据。

void TcpConnection::handleRead(Timestamp receiveTime)
{int savedErrno = 0;ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);if (n > 0){// 已建立连接的用户，有可读事件发生了，调用用户传入的回调操作onMessagemessageCallback_(shared_from_this(), &inputBuffer_, receiveTime);}else if (n == 0){handleClose();}else{errno = savedErrno;LOG_ERROR("TcpConnection::handleRead");handleError();}
}

TcpConnection::handleRead( )函数首先调用Buffer_.readFd(channel_->fd(), &saveErrno)，该函数底层调用Linux的函数readv( )，将Tcp接收缓冲区数据拷贝到用户定义的缓冲区中（inputBuffer_）。如果在读取拷贝的过程中发生了什么错误，这个错误信息就会保存在savedErrno中。
对于缓冲区 Buffer::readFd()函数之前的文章已经剖析过了。

二、TcpConnection的写

TcpConnection::send() 方法是用户调用发送接口，会调用TcpConnection::sendInLoop() 方法来处理具体的发送操作。如果在当前线程直接发送，就会调用 sendInLoop() 方法处理，否则需要把发送任务加入到事件循环中，等待对应的线程处理。

//给用户提供的 发送接口
void TcpConnection::send(const std::string &buf)
{if (state_ == kConnected){if (loop_->isInLoopThread()){// 判断当前的线程 是不是在对应的线程里面// 有一些情况sendInLoop(buf.c_str(), buf.size());}else{loop_->runInLoop(std::bind(&TcpConnection::sendInLoop,this,buf.c_str(),buf.size()));}}
}

在TcpConnection::sendInLoop() 方法的实现中接通过系统调用 write() 发送数据，如果有剩余未发送的数据，则会将数据添加到发送缓冲区中，并注册 channel 的可写事件，等待事件循环通知空闲后再进行发送。

/*** 发送数据  应用写的快， 而内核发送数据慢， 需要把待发送数据写入缓冲区，*  而且设置了水位回调*/ 
void TcpConnection::sendInLoop(const void* data, size_t len)
{ssize_t nwrote = 0;// remaining是没发送完的数据size_t remaining = len;bool faultError = false;// 之前调用过该connection的shutdown，不能再进行发送了if (state_ == kDisconnected){LOG_ERROR("disconnected, give up writing!");return;}// 表示channel_第一次开始写数据，而且缓冲区没有待发送数据if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0){// 返回的是具体发送的 数据nwrote = ::write(channel_->fd(), data, len);if (nwrote >= 0){remaining = len - nwrote;// 如果放松完了 ，并且注册了 发送完回调函数if (remaining == 0 && writeCompleteCallback_){// 既然在这里数据全部发送完成，就不用再给channel设置epollout事件了loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}}else // nwrote < 0{nwrote = 0;if (errno != EWOULDBLOCK){LOG_ERROR("TcpConnection::sendInLoop");if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE  RESET{faultError = true;}}}}// 说明当前这一次write，并没有把数据全部发送出去，// 剩余的数据需要保存到缓冲区当中，然后给channel// 注册epollout事件，poller发现tcp的发送缓冲区有空间，// 因为是lt模式  如果缓存区空余 就会不断地提醒// 会通知相应的sock-channel，调用writeCallback_回调方法也就是hanldwrite方法// 也就是调用TcpConnection::handleWrite方法，把发送缓冲区中的数据全部发送完成if (!faultError && remaining > 0) {// 目前发送缓冲区剩余的待发送数据的长度size_t oldLen = outputBuffer_.readableBytes();if (oldLen + remaining >= highWaterMark_&& oldLen < highWaterMark_&& highWaterMarkCallback_){loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), oldLen+remaining));}// 数据添加到缓冲区里面outputBuffer_.append((char*)data + nwrote, remaining);if (!channel_->isWriting()){// 这里一定要注册channel的写事件，否则poller不会给channel通知epolloutchannel_->enableWriting(); }}
}

发送缓冲区中有数据时，TcpConnection::handleWrite() 方法会被调用来处理具体的发送操作。在该方法中，首先会判断 channel 是否可写，如果可写则通过系统调用 writeFd() 将发送缓冲区中的数据写入到套接字中。如果写入成功，就会从发送缓冲区中删除已经发送的数据，并判断是否还有剩余数据，如果没有，则禁用 channel 的写事件，并执行可写回调函数。如果还有剩余数据，则会继续等待事件循环通知空闲后再次进行发送。

// 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{if (channel_->isWriting()){int savedErrno = 0;ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);if (n > 0){// 有数据发送成功  n个数据已经处理过了 把readable 向右移outputBuffer_.retrieve(n);if (outputBuffer_.readableBytes() == 0){// 已经发送完成了 编程不可写   执行回调写完回调writeCompleteCallback_channel_->disableWriting();if (writeCompleteCallback_){// 唤醒loop_对应的thread线程，执行回调// 唤醒线程 执行写完之后的回调事件loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}if (state_ == kDisconnecting){// 如果还有数据但是 就调用了 shutdown//   state_就变成了 == kDisconnecting// 但是 需要等待 数据传输完成 再调用shutdownInLoopshutdownInLoop();}}}else{LOG_ERROR("TcpConnection::handleWrite");}}else{LOG_ERROR("TcpConnection fd=%d is down, no more writing \n", channel_->fd());}
}

这里的细节问题就是如果想要关闭连接，那么通常是先关闭读端，等到将写缓冲区所有数据都写到tcp缓冲区后，再关闭写端，否则这些数据就不能发送给对端了

三、TcpConnection的关闭

需要关闭时候setState(kDisconnecting);把状态设置为kDisconnecting但是没有立即关闭，而是判断是否还有数据可写。

// 关闭连接
void TcpConnection::shutdown()
{if (state_ == kConnected){setState(kDisconnecting);loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));}
}
void TcpConnection::shutdownInLoop()
{// 如果buffer还有数据，这个就是writing状态// 会一直发， 知道发完 然后监控到状态是kDisconnecting // 再次调用这个函数 ，就会关闭了// 保证数据发送完if (!channel_->isWriting()) // 说明outputBuffer中的数据已经全部发送完成{socket_->shutdownWrite(); // 关闭写端}
}

如果buffer还有数据，这个就是writing状态会一直被epoll提醒发送，直到发完 然后监控到状态是kDisconnecting 再次调用这个函数 ，就会关闭了
保证数据发送完。

// 对outputBuffer_ 进行发送
void TcpConnection::handleWrite()
{if (channel_->isWriting()){int savedErrno = 0;ssize_t n = outputBuffer_.writeFd(channel_->fd(), &savedErrno);if (n > 0){// 有数据发送成功  n个数据已经处理过了 把readable 向右移outputBuffer_.retrieve(n);if (outputBuffer_.readableBytes() == 0){// 已经发送完成了 编程不可写   执行回调写完回调writeCompleteCallback_channel_->disableWriting();if (writeCompleteCallback_){// 唤醒loop_对应的thread线程，执行回调// 唤醒线程 执行写完之后的回调事件loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));}if (state_ == kDisconnecting){// 如果还有数据但是 就调用了 shutdown//   state_就变成了 == kDisconnecting// 但是 需要等待 数据传输完成 再调用shutdownInLoopshutdownInLoop();}}}else{LOG_ERROR("TcpConnection::handleWrite");}}else{LOG_ERROR("TcpConnection fd=%d is down, no more writing \n", channel_->fd());}
}