TcpConnection的读写操作【深度剖析】

2023-10-04 08:15

本文主要是介绍TcpConnection的读写操作【深度剖析】,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 前言
  • 一、TcpConnection的读
  • 二、TcpConnection的写
  • 三、TcpConnection的关闭


前言

今天总结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());}
}

这篇关于TcpConnection的读写操作【深度剖析】的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1669

相关文章

深度解析Java DTO(最新推荐)

《深度解析JavaDTO(最新推荐)》DTO(DataTransferObject)是一种用于在不同层(如Controller层、Service层)之间传输数据的对象设计模式,其核心目的是封装数据,... 目录一、什么是DTO?DTO的核心特点:二、为什么需要DTO?(对比Entity)三、实际应用场景解析

深度解析Java项目中包和包之间的联系

《深度解析Java项目中包和包之间的联系》文章浏览阅读850次,点赞13次,收藏8次。本文详细介绍了Java分层架构中的几个关键包:DTO、Controller、Service和Mapper。_jav... 目录前言一、各大包1.DTO1.1、DTO的核心用途1.2. DTO与实体类(Entity)的区别1

深度解析Python装饰器常见用法与进阶技巧

《深度解析Python装饰器常见用法与进阶技巧》Python装饰器(Decorator)是提升代码可读性与复用性的强大工具,本文将深入解析Python装饰器的原理,常见用法,进阶技巧与最佳实践,希望可... 目录装饰器的基本原理函数装饰器的常见用法带参数的装饰器类装饰器与方法装饰器装饰器的嵌套与组合进阶技巧

深度解析Spring Boot拦截器Interceptor与过滤器Filter的区别与实战指南

《深度解析SpringBoot拦截器Interceptor与过滤器Filter的区别与实战指南》本文深度解析SpringBoot中拦截器与过滤器的区别,涵盖执行顺序、依赖关系、异常处理等核心差异,并... 目录Spring Boot拦截器(Interceptor)与过滤器(Filter)深度解析:区别、实现

深度解析Spring AOP @Aspect 原理、实战与最佳实践教程

《深度解析SpringAOP@Aspect原理、实战与最佳实践教程》文章系统讲解了SpringAOP核心概念、实现方式及原理,涵盖横切关注点分离、代理机制(JDK/CGLIB)、切入点类型、性能... 目录1. @ASPect 核心概念1.1 AOP 编程范式1.2 @Aspect 关键特性2. 完整代码实

SpringBoot开发中十大常见陷阱深度解析与避坑指南

《SpringBoot开发中十大常见陷阱深度解析与避坑指南》在SpringBoot的开发过程中,即使是经验丰富的开发者也难免会遇到各种棘手的问题,本文将针对SpringBoot开发中十大常见的“坑... 目录引言一、配置总出错?是不是同时用了.properties和.yml?二、换个位置配置就失效?搞清楚加

Python中文件读取操作漏洞深度解析与防护指南

《Python中文件读取操作漏洞深度解析与防护指南》在Web应用开发中,文件操作是最基础也最危险的功能之一,这篇文章将全面剖析Python环境中常见的文件读取漏洞类型,成因及防护方案,感兴趣的小伙伴可... 目录引言一、静态资源处理中的路径穿越漏洞1.1 典型漏洞场景1.2 os.path.join()的陷

Spring Boot拦截器Interceptor与过滤器Filter深度解析(区别、实现与实战指南)

《SpringBoot拦截器Interceptor与过滤器Filter深度解析(区别、实现与实战指南)》:本文主要介绍SpringBoot拦截器Interceptor与过滤器Filter深度解析... 目录Spring Boot拦截器(Interceptor)与过滤器(Filter)深度解析:区别、实现与实

MyBatis分页插件PageHelper深度解析与实践指南

《MyBatis分页插件PageHelper深度解析与实践指南》在数据库操作中,分页查询是最常见的需求之一,传统的分页方式通常有两种内存分页和SQL分页,MyBatis作为优秀的ORM框架,本身并未提... 目录1. 为什么需要分页插件?2. PageHelper简介3. PageHelper集成与配置3.

Maven 插件配置分层架构深度解析

《Maven插件配置分层架构深度解析》:本文主要介绍Maven插件配置分层架构深度解析,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录Maven 插件配置分层架构深度解析引言:当构建逻辑遇上复杂配置第一章 Maven插件配置的三重境界1.1 插件配置的拓扑