首页 文章 精选 留言 我的

精选列表

搜索[零点击]，共10000篇文章
优秀的个人博客,低调大师

「从零单排canal 07」 parser模块源码解析

基于1.1.5-alpha版本，具体源码笔记可以参考我的github：https://github.com/saigu/JavaKnowledgeGraph/tree/master/code_reading/canal 本文将对canal的binlog订阅模块parser进行分析。 parser模块(绿色部分)在整个系统中的角色如下图所示，用来订阅binlog事件，然后通过sink投递到store。 parser模块应该来说是整个项目里面比较复杂的模块，代码非常多。 因此，本文根据过程中的主线来进行展开分析，从 启动 开始，进行分析。 如果读者有其他相关内容不明白的，可以给我留言，我会进行解答或者根据情况再单独写相关内容。 模块内的类如下： 重点需要关注几个核心问题 如何抓取binlog 对binlog消息处理做了怎样的性能优化 如何控制位点信息 如何兼容阿里云RDS的高可用模式下的主备切换问题 1.从启动进入parser主流程 前面的文章我们已经提到了，instance启动的是，会按照顺序启动instance的各个模块 parser模块就是在这里开始的。 这里需要注意一下，在beforeStartEventParser方法中，启动了parser的两个相关组件CanalLogPositionManager 和 CanalHAController，这里先分别介绍一下。 CanalLogPositionManager：管理位点信息 CanalHAController：instance连接源数据库的心跳检测，并实现数据库的HA（如果配置了standby的数据库） 1.1 位点信息管理CanalLogPositionManager 我们用的是default-instance.xml的配置，所以实际实现类是FailbackLogPositionManager 这里构造器有两个入参，一个是primary的MemoryLogPositionManager，一个是second的MetaLogPositionManager。 前者是内存的位点信息，后者我们我们看一下构造器的metaManager是基于zk的位点信息管理器。 所以FailbackLogPositionManager逻辑也比较简单，获取位点信息时，先尝试从内存memory中找到lastest position，如果不存在才尝试找一下zookeeper里的位点信息。 1.2 心跳控制器CanalHAController 我们用的是default-instance.xml的配置，所以实际实现类是HeartBeatHAController HeartBeatHAController里面没有特别复杂的逻辑，就是实现了心跳检测成功的onSuccess方法和onFail方法。另外维护了一个CanalHASwitchable对象，一旦心跳检测失败超过一定次数，就执行doSwitch()进行主备切换。 前提是我们要设置了主备数据库的连接信息。 这里的代码写的真的是有点混乱，居然是用MysqlEventParser实现了这个doSwitch()方法。 另外，在MysqlEventParser中，写了一个MysqlDetectingTimeTask内部类，集成了TimerTask来做定时心跳检测。 定时去连接数据库，可以通过select\desc\show\explain等方法做存活检测 如果检测成功，就调用HeartBeatHAController的onSuccess方法 如果失败，就HeartBeatHAController的onFail方法 如果失败超过一定次数，onFail方法中就会调用doSwitch方法进行主备切换 2.核心逻辑 从parser.start()进去后，我们就来到了parser的核心。 从default-instance.xml配置文件看，默认的parser实现是从base-instance.xml的baseEventParser来的，用的是RdsBinlogEventParserProxy类。 我们看下这个类图的结构。 从这个结构来看，我们从上到下，就能对parser模块的主体逻辑进行抽丝剥茧了。 Let’s go! 2.1 CanalEventParser接口 定义了一个空的接口 2.2 AbstractEventParser抽象类 这个类里面代码非常多，我们重点关注核心流程。 2.2.1 构造器AbstractEventParser() 构造器里面就只做了一件事情，创建了一个EventTransactionBuffer。 EventTransactionBuffer这个类顾名思义就是一个缓冲buffer，它的作用源码里的注释也很清楚，它是缓冲event队列，提供按事务刷新数据的机制。 那对于这里构造器中实现的TransactionFlushCallback的flush(List<CanalEntry.Entry> transaction) 方法，肯定就是对于事务中的一系列event，刷新到store中。 我们可以看下consumeTheEventAndProfilingIfNecessary(transaction)方法，跟我们想的一样，具体的sink方法放在后面的sink模块再展开分析。 2.2.2 主干的start()方法 主要做了这些事情： 初始化缓冲队列transactionBuffer 初始化binlogParser 启动一个新的线程进行核心工作 构造Erosa连接ErosaConnection 利用ErosaConnection启动一个心跳线程 执行dump前的准备工作,查看数据库的binlog_format和binlog_row_image，准备一下DatabaseTableMeta findStartPosition获取最后的位置信息（挺重要的，具体实现在MysqlEventParser） 构建一个sinkHandler，实现具体的sink逻辑（我们可以看到，里面就是把单个event事件写入到transactionBuffer中） 开始dump过程，默认是parallel处理的，需要构建一个MultiStageCoprocessor；如果不是parallel，就直接用sinkHandler处理。内部while不断循环，根据是否parallel，选择MultiStageCoprocessor或者sinkHandler进行投递。 如果有异常抛出，那么根据异常类型做相关处理，然后退出sink消费，释放一下状态，sleep一段时间后重新开始 代码很长，逻辑比较清晰，就不贴了。 2.2.3 核心dump过程 dump过程默认是parallel处理的，需要构建一个MultiStageCoprocessor；如果不是parallel，就直接用sinkHandler处理。内部while不断循环，根据是否parallel，选择MultiStageCoprocessor或者sinkHandler进行投递。 注意multiStageCoprocessor在这里start启动。 代码如下 dump方法在MysqlConnection类中实现，主要就是把自己注册到数据库作为一个slave，然后获取binlog变更，具体到协议我们就不展开分析了。 通过fetcher抓取到event，然后调用sink投递到store。 注意，parallel为false的，是单线程交给sinkHandler处理，parallel为true的，交给MultiStageCoprocessor的coprocessor.publish(buffer)处理，后面展开分析下并行处理的逻辑。 注意multiStageCoprocessor在这里publish进行写入RingBuffer，下文会详细讲下这里的机制。 2.3 AbstractMysqlEventParser抽象类 这个类比较简单，就是做了根据配置做了一些对象创建和设置的工作，比如BinlogParser的构建、filter的设置等 2.4 MysqlEventParser实现类 总共有将近1000行代码，里面其实代码组织有点混乱。像前面提到的MysqlDetectingTimeTask内部类、HeartBeatHAController的部分方法实现，都是在这个类里面的。 那抛开这些来说，这个类的主要功能还是在处理根据journalName、position、timestamp等配置查找对应的binlog位点。 我们选取核心流程里面的关键逻辑 findStartPostion( ) 方法进行分析即可。 这个是AbstractEventParser类中start方法中调用的，获取dump起始位点。 我们默认是使用 非GTID mode记录位点信息的，所以直接看下来看下findStartPositionInternal( ) 具体逻辑，这里可以了解到如何正确配置位点信息： logPositionManager找历史记录 如果没有找到 如果instance没有配置canal.instance.master.journal.name 如果instance配置了canal.instance.master.timestamp,就按照时间戳去binlog查找 如果没有配置timestamp，就返回数据库binlog最新的位点 如果instance配置了canal.instance.master.journal.name 如果instance配置了canal.instance.master.position，那就根据journalName和position获取bingo位点信息 如果配置了timestamp，就用journalName + timestamp形式获取位点信息 如果找到了历史记录 如果历史记录的连接信息和当前连接信息一致，那么判断下是否有异常，没有异常就直接返回 如果历史记录的连接信息和当前连接信息不一致，说明可能发生主备切换，就把历史记录的时间戳回退一分钟，重新查询 这里是纯if else 流程代码，挺长的，就不贴了。 在这个过程中，调用了几个有意思的方法，可以了解一下 findServerId( )：查询当前db的serverId信息，mysql命令为 show variables like 'server_id' findEndPosition（）：查询当前的binlog位置，mysql命令为 show master status findStartPosition（）：查询当前的binlog位置，mysql命令为 show binlog events limit 1 findSlavePosition（）：查询当前的slave视图的binlog位置，mysql命令为 show slave status 2.5 RdsBinlogEventParserProxy实现类 这个类比较简单，就是canal为阿里云rds定制的一个代理实现类。 主要解决了云rds本身高可用架构下，服务端HA切换后导致的binlog位点信息切换。 所以对于抛出的异常做了一定的处理，兼容了这种服务端HA的情况。 同时，也能满足rds的备份文件指定位点开始增量消费的特性。 主要过程如下 如果抛出了PositionNotFoundException异常，就委托rdsLocalBinlogEventParser进行处理 rdsLocalBinlogEventParser会通过下载binlog的oss备份，找到目标binlog文件和位置 3.事件处理优化 MultiStageCoprocessor 我们前面说过，parallel为false的，是单线程交给sinkHandler处理，parallel为true的，交给MultiStageCoprocessor处理。 这里展开看看并行处理是如何实现的。 实现类是MysqlMultiStageCoprocessor， 看下基本结构，持有了EventTransactionBuffer（前文提到过的存储事务内多个evnet的buffer）、RingBuffer<MessageEvent>、几个线程池、两个BatchEventProcessor<MessageEvent>。 这些属性类型基本是跟Disruptor框架相关。 start（）方法里面对一系列属性做了初始化配置并进行启动，要理解这里的逻辑，其实主要是使用Disruptor框架做的任务队列。 如果了解了Disruptor框架的使用，就能明白这里所做的任务队列处理模型了。 start（）源码如下： 首先，这里用了Disruptor框架的典型单生产者-多消费者模型。 这里创建生产者的时候，就创建了RingBuffer和Sequencer，全局唯一。 上面在dump方法内，订阅到binlog事件后，通过multiStageCoprocessor的publish方法写入RingBuffer，作为单一的生产者。 多消费者主要通过Disruptor的Sequencer管理。 Sequencer 接口有两种实现，SingleProducerSequencer 和 MultiProducerSequencer，分别来处理单个生产者和多个生产者的情况，这里使用了SingleProducerSequencer。 在 Sequencer 中有一个 next() 方法，就是这个方法来产生 Sequence 中的 value。Sequence 本质上可以认为是一个 AtomicLong，消费者和生产者都会维护自己的 Sequence。 Sequencer 的核心就是解决了两个问题，第一个是对于所有的消费者，在 RingBuffer 为空时，就不能再从中取数据，对于生产者，新生产的内容不能把未消费的数据覆盖掉。 上图中 C 代表消费者，P 代表生产者。 当然，在多消费者模型中，一个关键的问题是控制消费者的消费顺序。 这里主要通过消费者之间控制依赖关系其实就是控制 sequence 的大小，如果说 C2 消费者 依赖 C1，那就表示 C2 中 Sequence 的值一定小于等于 C1 的 Sequence。 具体的方法是通过RingBuffer的addGatingSequences( )进行的。 具体Disruptor的原理和使用就不展开说明了，这里了解这些关键问题即可。 通过这样的编程模型，parser实现了解析器的多阶段顺序处理。 Stage1: 网络接收 (单线程)，publish投递到RingBuffer Stage2: 从RingBuffe获取事件，使用SimpleParserStage进行基本解析 (单线程，事件类型、DDL解析构造TableMeta、维护位点信息) Stage3: 事件深度解析 ，用workpool进行多线程, 使用DmlParserStage进行DML事件数据的完整解析 Stage4: SinkStoreStage单线程投递到store SimpleParserStage和SinkStoreStage使用了stageExecutor这个线程池进行管理，DmlParserStage使用了workpool进行管理。 这三个类都是MysqlMultiStageCoprocessor的内部类，通过实现OnEvent方法进行逻辑处理，具体处理逻辑就不展开了，大家有兴趣可以看下源码。 4.总结 这个模块是非常核心的，涉及到了对binlog事件的抓取和处理，以及相关位点信息的处理。 回头看开头几个问题，相信也都有了答案： 如何抓取binlog dump方法在MysqlConnection类中实现，主要就是把自己注册到数据库作为一个slave，然后获取binlog变更（具体的协议我们就不展开分析了）。 对binlog消息处理做了怎样的性能优化 利用disruptor框架，基于RingBuffer实现了 单线程接受 -> 单线程解析事件 -> 多线程深度解析事件 -> 单线程投递store 这样的一个流程。 （这里有点疑惑，单线程接受事件后，为什么需要一个单线程先解析一下再多线程深度解析，而不是直接多线程深度解析？有了解的朋友可以给我留言指点一下，谢谢） 如何控制位点信息 有多种CanalLogPositionManager可以选择。 默认采用FailbackLogPositionManager，获取位点信息时，先尝试从内存memory中找到lastest position，如果不存在才尝试找一下zookeeper里的位点信息。 如何兼容阿里云RDS的高可用模式下的主备切换问题 RdsBinlogEventParserProxy如果发现了PositionNotFoundException异常，就委托rdsLocalBinlogEventParser通过下载binlog的oss备份，找到目标binlog文件和位置。 原创：阿丸笔记（微信公众号：aone_note），欢迎分享，转载请保留出处。 扫描下方二维码可以关注我哦～ 觉得不错，就点个 再看吧👇 本文分享自微信公众号 - 阿丸笔记（aone_note）。如有侵权，请联系 support@oschina.cn 删除。本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
2020-08-21
优秀的个人博客,低调大师

从零开始理解HTTP协议及报文分析

前言 从事性能测试必不可绕过的就是协议，对基本知识的了解，还是深入掌握协议的机制，都能让你在从事性能测试实施时显得更加顺手。 下面我们就HTTP协议及性能测试过程必须掌握的一些分析工具来进行分享。 重点分享性能测试实施过程中必须掌握的关键技术、工具。更细节的请参考HTTP相关书籍或RFC文档。 HTTP基本架构 下面我们用一张简单的流程图来展示HTTP协议基本架构，以便大家先有个基本的了解。 Web Client可以是浏览器、搜索引擎、机器人等等一切基于HTTP协议发起http请求的工具。 Web Server可以是任何的能解析HTTP请求，并返回给Web Client可识别的响应的服务，常见的有apache、nginx、IIS等等web服务器。 浓缩就是精华，看下最简洁的HTTP交互图： HTTP报文结构 请求报文 HTTP请求报文由请求行、请求头、空行和请求内容4个部分构成。 如下图所示： 下面对上图进行简单的分析： 请求行 由请求方法字段、URL字段、协议版本字段三部分构成，它们之间由空格隔开。常用的请求方法有：GET、POST、HEAD、PUT、DELETE、OPTIONS、TRACE、CONNECT。 请求头 请求头由key/value对组成，每行为一对，key和value之间通过冒号(:)分割。请求头的作用主要用于通知服务端有关于客户端的请求信息。 典型的请求头有： User-Agent：生成请求的浏览器类型 Accept：客户端可识别的响应内容类型列表；星号*用于按范围将类型分组。*/*表示可接受全部类型，type/*表示可接受type类型的所有子类型。 Accept-Language: 客户端可接受的自然语言 Accept-Encoding: 客户端可接受的编码压缩格式 Accept-Charset：可接受的字符集 Host: 请求的主机名，允许多个域名绑定同一IP地址 connection：连接方式（close或keeplive） Cookie: 存储在客户端的扩展字段 空行 最后一个请求头之后就是空行，用于告诉服务端以下内容不再是请求头的内容了。 请求内容 请求内容主要用于POST请求，与POST请求方法配套的请求头一般有Content-Type（标识请求内容的类型）和Content-Length（标识请求内容的长度） 响应报文 HTTP响应报文由状态行、响应头、空行和响应内容4个部分构成。 如下图所示： 下面对响应报文格式进行简要的分析说明： 状态行 由HTTP协议版本、状态码、状态码描述三部分构成，它们之间由空格隔开。 状态码由3位数字组成，第一位标识响应的类型，常用的5大类状态码如下： 1xx：表示服务器已接收了客户端的请求，客户端可以继续发送请求 2xx：表示服务器已成功接收到请求并进行处理 3xx：表示服务器要求客户端重定向 4xx：表示客户端的请求有非法内容 5xx：标识服务器未能正常处理客户端的请求而出现意外错误 常见状态码说明： 200 OK：表示客户端请求成功 400 Bad Request：表示客户端请求有语法错误，不能被服务器端解析 401 Unauthonzed：表示请求未经授权，该状态码必须与WWW-Authenticate报文头一起使用 404 Not Found：请求的资源不存在，例如输入了错误的url 500 Internal Server Error：表示服务器发生了不可预期的错误，导致无法完成客户端的请求 503 Service Unavailable：表示服务器当前不能处理客户端的请求，在一段时间后服务器可能恢复正常 响应头 一般情况下，响应头会包含以下，甚至更多的信息。 Location：服务器返回给客户端，用于重定向到新的位置 Server：包含服务器用来处理请求的软件信息及版本信息 Vary：标识不可缓存的请求头列表 Connection: 连接方式。 对于请求端来讲：close是告诉服务端，断开连接，不用等待后续的求请了。keeplive则是告诉服务端，在完成本次请求的响应后，保持连接，等待本次连接后的后续请求。 对于响应端来讲：close表示连接已经关闭。keeplive则表示连接保持中，可以继续处理后续请求。Keep-Alive表示如果请求端保持连接，则该请求头部信息表明期望服务端保持连接多长时间（秒），例如300秒，应该这样写Keep-Alive: 300 空行 最后一个响应头之后就是空行，用于告诉请求端以下内容不再是响应头的内容了。 响应内容 服务端返回给请求端的文本信息。 HTTP报文示例 在这里我们在Firefox下用firebug随意抓取一个HTTP包和上文的报文结构做下一一对应关系图，以便大家了解实际的包和标准报文结构的对应关系。 总结 对于HTTP协议的交互过程这里就不再进行说明了，大家可以搜索下相关的资料进行学习，上述的内容请务必熟练掌握、深刻了解。更详细的内容推荐大家学习RFC 2616（http协议1.1版本，有中文版本） 更多精彩： 工具篇 - JMeter之旅01 工具篇 - JMeter目录及关键配置分析02 webdriver介绍&与Selenium RC的比较 Python Selenium Webdriver安装手册 创建你的第一个webdriver python代码 基于Excel参数化你的Selenium2测试 Python Selenium设计模式-POM Python多线程Selenium跨浏览器测试 python selenium2示例 - 生成 HTMLTestRunner 测试报告 随意支持下，一起做好分享 扫一扫或长按关注开源优测公众号，原创文章首发平台 在公众号里回复一下关键字获取对应的系列文章 物联网 appium 大数据测试 RobotFramework Python 自动驾驶 jmeter selenium jenkins 接口测试 顺手点点点右下角的 本文分享自微信公众号 - 开源优测（DeepTest）。如有侵权，请联系 support@oschina.cn 删除。本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
2020-08-12

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2026-04-11 大小: 211.28KB 下载: 56次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2026-09-24 大小: 189MB 下载: 9次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-08-21 大小: 1.42GB 下载: 6次
WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。

时间: 2025-12-13 大小: 249.54MB 下载: 22次
热门标签
DeepSeek Jdk25 OpenAi HarmonyOS6 Nacos3 SpringBoot4 SpringCloud Docker Kubernetes Service Mesh Redis Gemini3 Rocky

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册