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的读写操作【深度剖析】
文章目录 前言一、TcpConnection的读二、TcpConnection的写三、TcpConnection的关闭 前言 今天总结TcpConnection类的读写事件。 一、TcpConnection的读 当Poller检测到套接字的Channel处于可读状态时,会调用Channel的回调函数,回调函数中根据不同激活…...
k8s面试题
1 简述etcd及其特点 ETCD是高可用分布式的键值存储系统 特点 1)强一致性:即使部分节点故障,etcd仍能正常工作,并保持数据一致 (强一致性:分布式系统中,更新操作,所有节点读取的数据都是最新的,一致的) 2)高可用:etcd支持数据多副本复制,一个节点挂,其他节点接…...
OpenCV 4.x 版本的新特性都有哪些?
文章大纲 V 4. 0DNNV 4.0 - 4. 8cheatsheetvideo analysis参考文献与学习路径2016年的时候我快研究生毕业了,那时候OpenCV 2.4.x 版本非常的流行,当时3.x 的版本刚发布,很多人都没有用习惯。 我写过一遍笔记: OpenCV 3.0 3.1版本的改进家里还有一本书:《OpenCV 3 计算机视…...
Redisson—分布式集合
7.1. 映射(Map) 基于Redis的Redisson的分布式映射结构的RMap Java对象实现了java.util.concurrent.ConcurrentMap接口和java.util.Map接口。与HashMap不同的是,RMap保持了元素的插入顺序。该对象的最大容量受Redis限制,最大元素数…...

93、Redis 之 使用连接池管理Redis6.0以上的连接 及 消息的订阅与发布
★ 使用连接池管理Redis连接 从Redis 6.0开始,Redis可支持使用多线程来接收、处理客户端命令,因此应用程序可使用连接池来管理Redis连接。 上一章讲的是创建单个连接来操作redis数据库,这次使用连接池来操作redis数据库 Lettuce连接池 支持…...
doris动态分区开启历史分区
举例说明: CREATE TABLE tbl1 (k1 DATE,... ) PARTITION BY RANGE(k1) () DISTRIBUTED BY HASH(k1) PROPERTIES ("dynamic_partition.enable" "true","dynamic_partition.time_unit" "DAY","dynamic_partition.sta…...

Linux用户与权限(认知root用户、修改权限控制 - chmod、修改权限控制 - chown)
目录 1. 认知root用户 1.1 什么是root用户(超级管理员) 1.2 用户切换命令 1.3 sudo命令 1.3.1 为普通用户配置sudo认证 2. 用户、用户组管理 2.1 理解用户、用户组的概念 2.2 掌握用户、用户组管理的相关命令 2.2.1 用户组管理 2.2.2 …...

处理conda安装工具的动态库问题——解决记录 libssl.1.0.0 系统中所有openssl位置全览 whereis openssl
处理conda安装工具的动态库问题——解决记录 处理conda安装工具的动态库问题——解决记录 - 简书 解决libssl.so.1.0.0: cannot open shared object file: No such file or directory问题 - 简书 openssl 默认版本问题(Anaconda相关)_anaconda openssl-…...
如何在Go中格式化字符串
由于字符串通常由书面文本组成,在很多情况下,我们可能希望通过标点符号、换行和缩进来更好地控制字符串的外观,以使其更易于阅读。 在本教程中,我们将介绍一些处理go字符串的方法,以确保所有输出文本的格式正确。 字…...

C程序设计内容与例题讲解 -- 第四章--选择结构程序设计第二部分(第五版)谭浩强
前言:在前面我们学习了选择结构和条件判断,用if语句实现选择结构,关系运算符和关系表达式,逻辑运算符和逻辑表达式等知识。今天我们将接着上一篇未讲完的继续讲解。 鸡汤:种一棵树最好的时间是十年前,其次是现在!加油各…...

接雨水问题
接雨水问题 问题背景 LeetCode 42. 接雨水 接雨水问题是一个经典的计算雨水滞留量的问题,通常使用柱状图来表示不同高度的柱子。在下雨的情况下,柱子之间的凹陷部分能够存储雨水,问题的目标是计算这些柱子所能接收的雨水总量。 相关知识 …...

小谈设计模式(9)—工厂方法模式
小谈设计模式(9)—工厂方法模式 专栏介绍专栏地址专栏介绍 工厂方法模式角色分类抽象产品(Abstract Product)具体产品(Concrete Product)抽象工厂(Abstract Factory)具体工厂&#x…...

Android etc1tool之png图片转换pkm 和 zipalign简介
关于作者:CSDN内容合伙人、技术专家, 从零开始做日活千万级APP。 专注于分享各领域原创系列文章 ,擅长java后端、移动开发、商业变现、人工智能等,希望大家多多支持。 目录 一、导读二、etc1tool2.1、用法 三、zipalign3.1 使用 四…...
Spring Boot快速入门:构建简单的Web应用
SpringBoot Spring Boot是一个用于简化Spring应用程序开发的框架,它通过提供开箱即用的配置和一组常用的功能,使得构建高效、可维护的应用变得非常容易。在本篇博客中,我们将一步步地介绍如何快速入门Spring Boot,并构建一个简单的…...
JAVA 泛型、序列化和复制
泛型提供了编译时类型安全检测机制,该机制允许程序员在编译时检测到非法的类型。泛型的本质是参数化类型,也就是说所操作的数据类型被指定为一个参数。比如我们要写一个排序方法,能够对整型数组、字符串数组甚至其他任何类型的数组进行排序&a…...

以太网基础学习(二)——ARP协议
一、什么是MAC地址 MAC地址(英语:Media Access Control Address),直译为媒体访问控制位址,也称为局域网地址(LAN Address),MAC位址,以太网地址(Ethernet Addr…...

【Java-LangChain:使用 ChatGPT API 搭建系统-4】评估输入-分类
第三章,评估输入-分类 如果您正在构建一个允许用户输入信息的系统,首先要确保人们在负责任地使用系统,以及他们没有试图以某种方式滥用系统,这是非常重要的。 在本章中,我们将介绍几种策略来实现这一目标。 我们将学习…...

嵌入式Linux应用开发-驱动大全-第一章同步与互斥③
嵌入式Linux应用开发-驱动大全-第一章同步与互斥③ 第一章 同步与互斥③1.4 Linux锁的介绍与使用1.4.1 锁的类型1.4.1.1 自旋锁1.4.1.2 睡眠锁 1.4.2 锁的内核函数1.4.2.1 自旋锁1.4.2.2 信号量1.4.2.3 互斥量1.4.2.4 semaphore和 mutex的区别 1.4.3 何时用何种锁1.4.4 内核抢占…...

树的存储结构以及树,二叉树,森林之间的转换
目录 1.双亲表示法 2.孩子链表 3.孩子兄弟表示法 4.树与二叉树的转换 (1)树转换为二叉树 (2)二叉树转换成树 5.二叉树与森林的转化 (1)森林转换为二叉树 以下树为例 1.双亲表示法 双亲表示法定义了…...

【AI视野·今日NLP 自然语言处理论文速览 第四十二期】Wed, 27 Sep 2023
AI视野今日CS.NLP 自然语言处理论文速览 Wed, 27 Sep 2023 Totally 50 papers 👉上期速览✈更多精彩请移步主页 Daily Computation and Language Papers Attention Satisfies: A Constraint-Satisfaction Lens on Factual Errors of Language Models Authors Mert …...
[特殊字符] 智能合约中的数据是如何在区块链中保持一致的?
🧠 智能合约中的数据是如何在区块链中保持一致的? 为什么所有区块链节点都能得出相同结果?合约调用这么复杂,状态真能保持一致吗?本篇带你从底层视角理解“状态一致性”的真相。 一、智能合约的数据存储在哪里…...

【大模型RAG】拍照搜题技术架构速览:三层管道、两级检索、兜底大模型
摘要 拍照搜题系统采用“三层管道(多模态 OCR → 语义检索 → 答案渲染)、两级检索(倒排 BM25 向量 HNSW)并以大语言模型兜底”的整体框架: 多模态 OCR 层 将题目图片经过超分、去噪、倾斜校正后,分别用…...

iOS 26 携众系统重磅更新,但“苹果智能”仍与国行无缘
美国西海岸的夏天,再次被苹果点燃。一年一度的全球开发者大会 WWDC25 如期而至,这不仅是开发者的盛宴,更是全球数亿苹果用户翘首以盼的科技春晚。今年,苹果依旧为我们带来了全家桶式的系统更新,包括 iOS 26、iPadOS 26…...
C++:std::is_convertible
C++标志库中提供is_convertible,可以测试一种类型是否可以转换为另一只类型: template <class From, class To> struct is_convertible; 使用举例: #include <iostream> #include <string>using namespace std;struct A { }; struct B : A { };int main…...

Psychopy音频的使用
Psychopy音频的使用 本文主要解决以下问题: 指定音频引擎与设备;播放音频文件 本文所使用的环境: Python3.10 numpy2.2.6 psychopy2025.1.1 psychtoolbox3.0.19.14 一、音频配置 Psychopy文档链接为Sound - for audio playback — Psy…...

微信小程序云开发平台MySQL的连接方式
注:微信小程序云开发平台指的是腾讯云开发 先给结论:微信小程序云开发平台的MySQL,无法通过获取数据库连接信息的方式进行连接,连接只能通过云开发的SDK连接,具体要参考官方文档: 为什么? 因为…...
使用Matplotlib创建炫酷的3D散点图:数据可视化的新维度
文章目录 基础实现代码代码解析进阶技巧1. 自定义点的大小和颜色2. 添加图例和样式美化3. 真实数据应用示例实用技巧与注意事项完整示例(带样式)应用场景在数据科学和可视化领域,三维图形能为我们提供更丰富的数据洞察。本文将手把手教你如何使用Python的Matplotlib库创建引…...
现有的 Redis 分布式锁库(如 Redisson)提供了哪些便利?
现有的 Redis 分布式锁库(如 Redisson)相比于开发者自己基于 Redis 命令(如 SETNX, EXPIRE, DEL)手动实现分布式锁,提供了巨大的便利性和健壮性。主要体现在以下几个方面: 原子性保证 (Atomicity)ÿ…...

【从零学习JVM|第三篇】类的生命周期(高频面试题)
前言: 在Java编程中,类的生命周期是指类从被加载到内存中开始,到被卸载出内存为止的整个过程。了解类的生命周期对于理解Java程序的运行机制以及性能优化非常重要。本文会深入探寻类的生命周期,让读者对此有深刻印象。 目录 …...
多模态图像修复系统:基于深度学习的图片修复实现
多模态图像修复系统:基于深度学习的图片修复实现 1. 系统概述 本系统使用多模态大模型(Stable Diffusion Inpainting)实现图像修复功能,结合文本描述和图片输入,对指定区域进行内容修复。系统包含完整的数据处理、模型训练、推理部署流程。 import torch import numpy …...