当前位置： 首页 > news >正文

MySql-MySqlConnector

news 来源：原创 2024/9/19 21:24:43

提示：MySqlConnector 类的主要职责是从MySQL数据库中捕获数据变更，并将这些变更以事件的形式发布到Kafka中。这使得下游的应用程序可以通过订阅Kafka主题来实时获取MySQL数据库中的变更信息。

文章目录

  • 前言
  • 一、核心功能
  • 二、代码分析
  • 总结

前言

提示：MySqlConnector 类是一个关键组件，它负责设置和管理整个数据捕获流程，从配置到数据库连接，再到数据变更事件的捕获和发送。这对于实现实时数据同步和流处理是非常重要的。

提示：以下是本篇文章正文内容

一、核心功能

核心功能详细说明

  1. 数据变更捕获：

    • 通过读取 MySQL 的二进制日志 (binlog) 来捕获数据库中的数据变更事件，包括插入、更新和删除等操作。

  2. Kafka Connect 兼容性：

    • 实现了 Kafka Connect 的接口，允许该连接器与 Kafka Connect 平滑集成。
    • 提供了 taskClass() 方法返回任务类 MySqlConnectorTask，这是实际执行数据捕获工作的类。

  3. 配置管理：

    • 通过 config() 方法返回配置定义 (ConfigDef)，这些配置定义了连接器运行所需的参数。
    • 使用 MySqlConnectorConfig 类来管理配置选项。

  4. 版本信息：

    • 通过 version() 方法提供连接器的版本信息。

  5. 连接器任务创建：

    • 通过 taskClass() 方法指定任务类，即 MySqlConnectorTask，这是执行数据捕获的具体任务类。

  6. 配置验证：

    • 通过 validateAllFields() 方法对配置进行验证，确保所有必需的字段都已正确设置。

  7. 数据库连接建立：

    • 通过 createConnection() 方法建立到 MySQL 数据库的实际连接。
    • 使用 MySqlConnection 和 MySqlConnectionConfiguration 来配置和管理数据库连接。

  8. 连接器配置创建：

    • 通过 createConnectorConfig() 方法创建并返回 MySqlConnectorConfig 实例，该实例包含了连接器运行所需的配置信息。

二、代码分析

package io.debezium.connector.mysql;import java.util.Map;import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigDef.Width;
import org.apache.kafka.common.config.ConfigDef.ValidString;
import org.apache.kafka.common.config.ConfigDef.ValidList;
import org.apache.kafka.common.config.ConfigDef.ValidBoolean;
import org.apache.kafka.common.config.ConfigDef.ValidInt;
import org.apache.kafka.common.config.ConfigDef.ValidLong;
import org.apache.kafka.common.config.ConfigDef.ValidDouble;
import org.apache.kafka.common.config.ConfigDef.ValidDuration;
import org.apache.kafka.common.config.ConfigDef.ValidBytesize;
import org.apache.kafka.common.config.ConfigDef.ValidPort;
import org.apache.kafka.common.config.ConfigDef.ValidRegex;
import org.apache.kafka.common.config.ConfigDef.ValidEnum;
import org.apache.kafka.common.config.ConfigDef.ValidSymbolic;
import org.apache.kafka.common.config.ConfigDef.ValidPassword;
import org.apache.kafka.common.config.ConfigDef.ValidPath;
import org.apache.kafka.common.config.ConfigDef.ValidUrl;
import org.apache.kafka.common.config.ConfigDef.ValidJson;
import org.apache.kafka.common.config.ConfigDef.ValidJsonArray;
import org.apache.kafka.common.config.ConfigDef.ValidJsonMap;
import org.apache.kafka.common.config.ConfigDef.ValidPattern;
import org.apache.kafka.common.config.ConfigDef.ValidClass;
import org.apache.kafka.common.config.ConfigDef.ValidScript;
import org.apache.kafka.common.config.ConfigDef.ValidExpression;
import org.apache.kafka.common.config.ConfigDef.ValidTimestamp;
import org.apache.kafka.common.config.ConfigDef.ValidDate;
import org.apache.kafka.common.config.ConfigDef.ValidTime;
import org.apache.kafka.common.config.ConfigDef.ValidDurationOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidDurationOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidDurationOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidBytesizeOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidBytesizeOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidBytesizeOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidIntOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidIntOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidIntOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidLongOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidLongOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidLongOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidDoubleOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidDoubleOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidDoubleOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidFloatOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidFloatOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidFloatOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidShortOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidShortOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidShortOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidByteOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidByteOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidByteOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidCharOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidCharOrNegative;
import org.apache.kafka.common.config.ConfigDef.ValidCharOrPositive;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrZero;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrOne;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrTrue;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrFalse;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrOn;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrOff;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrYes;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrNo;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrEnabled;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrDisabled;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrTrueFalse;
import org.apache.kafka.common.config.ConfigDef.ValidBooleanOrOnOff;/*** A Kafka Connect source connector that creates tasks that read the MySQL binary log and generating the corresponding* data change events.* <h2>Configuration</h2>* <p>* This connector is configured with the set of properties described in {@link MySqlConnectorConfig}.*** @author Randall Hauch*/
public class MySqlConnector extends BinlogConnector<MySqlConnectorConfig> {// 定义了一个名为 MySqlConnector 的类，继承自 BinlogConnector，用于从 MySQL 数据库中捕获数据变更事件。public MySqlConnector() {// 构造函数。}@Overridepublic String version() {return Module.version();}// 返回当前连接器的版本信息。@Overridepublic Class<? extends Task> taskClass() {return MySqlConnectorTask.class;}// 返回任务类，即执行数据捕获任务的具体类。@Overridepublic ConfigDef config() {return MySqlConnectorConfig.configDef();}// 返回配置定义，定义了连接器运行所需的配置项。@Overrideprotected Map<String, ConfigValue> validateAllFields(Configuration config) {return config.validate(MySqlConnectorConfig.ALL_FIELDS);}// 验证配置项是否有效，确保所有必需的字段都已正确设置。@Overrideprotected MySqlConnection createConnection(Configuration config, MySqlConnectorConfig connectorConfig) {return new MySqlConnection(new MySqlConnectionConfiguration(config),MySqlFieldReaderResolver.resolve(connectorConfig));}// 创建 MySQL 数据库连接。@Overrideprotected MySqlConnectorConfig createConnectorConfig(Configuration config) {return new MySqlConnectorConfig(config);}// 创建连接器配置实例。
}

类的设计与封装

MySqlConnector 类是一个很好的面向对象设计的例子。它通过继承 BinlogConnector 类实现了特定的功能，同时通过封装实现了对 MySQL 数据库的专有支持。

继承与多态

  1. 继承：MySqlConnector 继承自 BinlogConnector，这意味着它可以复用基类提供的通用功能，如连接器的基本生命周期管理等。这种设计减少了重复代码，并且使得维护更加容易。

  2. 多态：通过覆盖父类的方法（如 taskClass()、config() 等），MySqlConnector 能够提供针对 MySQL 特定的行为，同时也保持了与 Kafka Connect 框架的兼容性。

封装

  1. 配置管理：通过 MySqlConnectorConfig 类来管理配置，这使得配置的细节被封装起来，外部不需要关心配置的具体实现细节。

  2. 数据库连接：通过 createConnection() 方法创建数据库连接，这使得连接的创建过程被封装在类内部，外部只需要调用方法即可获得连接。

抽象与具体

  1. 抽象：BinlogConnector 类提供了一个抽象的基础框架，定义了连接器的基本行为。

  2. 具体：MySqlConnector 类则是具体的实现，它提供了针对 MySQL 数据库的具体支持，如配置的定制、数据库连接的建立等。

启发

  1. 模块化设计：通过继承和多态，我们可以很容易地扩展新的数据库连接器，只需继承 BinlogConnector 并覆盖必要的方法即可。

  2. 可维护性和可扩展性：通过将通用功能与特定实现分离，使得代码更容易维护和扩展。例如，如果需要添加对另一个数据库的支持，只需要创建一个新的子类即可。

代码优点

  1. 清晰的接口：MySqlConnector 类提供了清晰的方法签名，如 version()、taskClass() 和 config() 等，这使得其他开发者能够很容易地了解如何使用这个类。

  2. 良好的封装：通过将配置管理和数据库连接的创建封装在类内部，提高了代码的内聚性，降低了耦合度。

  3. 易于扩展：通过继承和多态，使得添加新的功能或支持新的数据库变得相对简单。

  4. 遵循设计模式：该类遵循了面向对象设计的原则，如单一职责原则、开放封闭原则等，这有助于提高代码的质量。

总结

MySqlConnector 类是 Debezium 项目的一部分，它作为一个 Kafka Connect 源连接器，其核心功能和作用如下：

  1. 数据变更捕获：

    • 从 MySQL 数据库的二进制日志 (binlog) 中捕获数据变更事件，包括插入、更新和删除等操作。

  2. Kafka Connect 兼容：

    • 实现了 Kafka Connect 的接口，允许该连接器与 Kafka Connect 平滑集成。
    • 提供了 taskClass() 方法返回任务类 MySqlConnectorTask，这是实际执行数据捕获工作的类。

  3. 配置管理：

    • 通过 config() 方法返回配置定义 (ConfigDef)，这些配置定义了连接器运行所需的参数。
    • 使用 MySqlConnectorConfig 类来管理配置选项。

  4. 版本信息：

    • 通过 version() 方法提供连接器的版本信息。

  5. 连接器任务创建：

    • 通过 taskClass() 方法指定任务类，即 MySqlConnectorTask，这是执行数据捕获的具体任务类。

  6. 配置验证：

    • 通过 validateAllFields() 方法对配置进行验证，确保所有必需的字段都已正确设置。

  7. 数据库连接建立：

    • 通过 createConnection() 方法建立到 MySQL 数据库的实际连接。
    • 使用 MySqlConnection 和 MySqlConnectionConfiguration 来配置和管理数据库连接。

  8. 连接器配置创建：

    • 通过 createConnectorConfig() 方法创建并返回 MySqlConnectorConfig 实例，该实例包含了连接器运行所需的配置信息。

MySqlConnector 类是一个关键组件，它负责设置和管理整个数据捕获流程，从配置到数据库连接，再到数据变更事件的捕获和发送。这对于实现实时数据同步和流处理是非常重要的。通过使用 MySqlConnector，用户可以轻松地将 MySQL 数据库中的数据变更以事件的形式发送到 Kafka 中，从而实现数据的实时处理和分析。

相关文章：

  • 北京网站建设多少钱？
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • docker 部署 mysql8
  • 设计模式-单一职责模式
  • WPS excel 数据最小二乘法拟合
  • 4、命令式和声明式
  • Linux shell编程：监控进程CPU使用率并使用 perf 抓取高CPU进程信息
  • 学习前端面试知识
  • cnetos部署高可用以及七层负载均衡
  • git clone 大文件 eof 错误
  • uniapp——列表选择样式
  • 消息队列项目
  • 职业本科大数据实训室
  • 如何使用 AWS CLI 创建和运行 EMR 集群
  • Nginx + PHP 8.0支持视频上传
  • Golang | Leetcode Golang题解之第326题3的幂
  • SoildWorks练习清单
  • 「前端早读君006」移动开发必备：那些玩转H5的小技巧
  • CentOS从零开始部署Nodejs项目
  • Facebook AccountKit 接入的坑点
  • HTTP 简介
  • IDEA 插件开发入门教程
  • JavaScript 事件——“事件类型”中“HTML5事件”的注意要点
  • Python爬虫--- 1.3 BS4库的解析器
  • Quartz实现数据同步 | 从0开始构建SpringCloud微服务(3)
  • RedisSerializer之JdkSerializationRedisSerializer分析
  • Spring Cloud Feign的两种使用姿势
  • 闭包--闭包作用之保存(一)
  • 如何合理的规划jvm性能调优
  • 如何借助 NoSQL 提高 JPA 应用性能
  • 远离DoS攻击 Windows Server 2016发布DNS政策
  • Hibernate主键生成策略及选择
  • 分布式关系型数据库服务 DRDS 支持显示的 Prepare 及逻辑库锁功能等多项能力 ...
  • ### RabbitMQ五种工作模式：
  • #进阶：轻量级ORM框架Dapper的使用教程与原理详解
  • #我与Java虚拟机的故事#连载01：人在JVM，身不由己
  • (16)UiBot：智能化软件机器人（以头歌抓取课程数据为例）
  • （2）STM32单片机上位机
  • （第61天）多租户架构（CDB/PDB）
  • （二）十分简易快速 自己训练样本 opencv级联lbp分类器 车牌识别
  • (计算机网络)物理层
  • （论文阅读11/100）Fast R-CNN
  • (免费领源码）python#django#mysql校园校园宿舍管理系统84831-计算机毕业设计项目选题推荐
  • (三)docker：Dockerfile构建容器运行jar包
  • （四）Linux Shell编程——输入输出重定向
  • （五）c52学习之旅-静态数码管
  • . ./ bash dash source 这五种执行shell脚本方式 区别
  • .Net 基于MiniExcel的导入功能接口示例
  • .net 使用$.ajax实现从前台调用后台方法(包含静态方法和非静态方法调用)
  • .NET/C# 解压 Zip 文件时出现异常：System.IO.InvalidDataException: 找不到中央目录结尾记录。
  • .NET/C# 异常处理：写一个空的 try 块代码，而把重要代码写到 finally 中（Constrained Execution Regions）
  • .NET构架之我见
  • .xml 下拉列表_RecyclerView嵌套recyclerview实现二级下拉列表，包含自定义IOS对话框...
  • @AliasFor注解
  • @GetMapping和@RequestMapping的区别
  • [ 代码审计篇 ] 代码审计案例详解(一) SQL注入代码审计案例
  • [20160902]rm -rf的惨案.txt