欢迎来到尧图网

客户服务 关于我们

您的位置：首页 > 汽车 > 新车 > 从源码出发:全面理解 Kafka Connect Jdbc与Kafka Connect 机制

从源码出发:全面理解 Kafka Connect Jdbc与Kafka Connect 机制

2025/6/22 2:14:13 来源：https://blog.csdn.net/qq_42773076/article/details/148691990  浏览:  次   关键词：从源码出发:全面理解 Kafka Connect Jdbc与Kafka Connect 机制

引言

kafka-connect-jdbc 是一个 Kafka Connector，用于在任何兼容 JDBC 的数据库和 Kafka 之间加载数据。它支持从数据库中读取数据并将其写入 Kafka 主题，也可以将 Kafka 主题中的数据写入数据库。

核心类与组件

1. JdbcSourceConnector
  • 功能：这是 Kafka Connect 的源连接器实现类，负责监控 JDBC 数据库并生成任务以摄取数据库内容。
  • 关键代码分析：
public class JdbcSourceConnector extends SourceConnector {// ... 省略部分代码@Overridepublic void start(Map<String, String> properties) throws ConnectException {log.info("Starting JDBC Source Connector");try {configProperties = properties;config = new JdbcSourceConnectorConfig(configProperties);} catch (ConfigException e) {throw new ConnectException("Couldn't start JdbcSourceConnector due to configuration error", e);}// 获取数据库连接final String dbUrl = config.getString(JdbcSourceConnectorConfig.CONNECTION_URL_CONFIG);final int maxConnectionAttempts = config.getInt(JdbcSourceConnectorConfig.CONNECTION_ATTEMPTS_CONFIG);final long connectionRetryBackoff = config.getLong(JdbcSourceConnectorConfig.CONNECTION_BACKOFF_CONFIG);dialect = DatabaseDialects.findBestFor(dbUrl, config);cachedConnectionProvider = connectionProvider(maxConnectionAttempts, connectionRetryBackoff);cachedConnectionProvider.getConnection();// 启动表监控线程long tablePollMs = config.getLong(JdbcSourceConnectorConfig.TABLE_POLL_INTERVAL_MS_CONFIG);long tableStartupLimitMs = config.getLong(JdbcSourceConnectorConfig.TABLE_MONITORING_STARTUP_POLLING_LIMIT_MS_CONFIG);List<String> whitelist = config.getList(JdbcSourceConnectorConfig.TABLE_WHITELIST_CONFIG);Set<String> whitelistSet = whitelist.isEmpty() ? null : new HashSet<>(whitelist);List<String> blacklist = config.getList(JdbcSourceConnectorConfig.TABLE_BLACKLIST_CONFIG);Set<String> blacklistSet = blacklist.isEmpty() ? null : new HashSet<>(blacklist);tableMonitorThread = new TableMonitorThread(dialect,cachedConnectionProvider,context,tableStartupLimitMs,tablePollMs,whitelistSet,blacklistSet,Time.SYSTEM);if (config.getString(JdbcSourceConnectorConfig.QUERY_CONFIG).isEmpty()) {tableMonitorThread.start();log.info("Starting Table Monitor Thread");}}@Overridepublic Class<? extends Task> taskClass() {return JdbcSourceTask.class;}@Overridepublic List<Map<String, String>> taskConfigs(int maxTasks) {String query = config.getString(JdbcSourceConnectorConfig.QUERY_CONFIG);List<Map<String, String>> taskConfigs;if (!query.isEmpty()) {// 自定义查询Map<String, String> taskProps = new HashMap<>(configProperties);taskProps.put(JdbcSourceTaskConfig.TABLES_CONFIG, "");taskProps.put(JdbcSourceTaskConfig.TABLES_FETCHED, "true");taskConfigs = Collections.singletonList(taskProps);} else {// 表模式List<TableId> currentTables = tableMonitorThread.tables();if (currentTables == null || currentTables.isEmpty()) {// 没有找到表taskConfigs = new ArrayList<>(1);Map<String, String> taskProps = new HashMap<>(configProperties);taskProps.put(JdbcSourceTaskConfig.TABLES_CONFIG, "");taskProps.put(JdbcSourceTaskConfig.TABLES_FETCHED, currentTables == null ? "false" : "true");taskConfigs.add(taskProps);} else {// 分配表到任务int numGroups = Math.min(currentTables.size(), maxTasks);List<List<TableId>> tablesGrouped = ConnectorUtils.groupPartitions(currentTables, numGroups);taskConfigs = new ArrayList<>(tablesGrouped.size());for (List<TableId> taskTables : tablesGrouped) {Map<String, String> taskProps = new HashMap<>(configProperties);ExpressionBuilder builder = dialect.expressionBuilder();builder.appendList().delimitedBy(",").of(taskTables);taskProps.put(JdbcSourceTaskConfig.TABLES_CONFIG, builder.toString());taskProps.put(JdbcSourceTaskConfig.TABLES_FETCHED, "true");taskConfigs.add(taskProps);}}}return taskConfigs;}@Overridepublic void stop() throws ConnectException {log.info("Stopping table monitoring thread");tableMonitorThread.shutdown();try {tableMonitorThread.join(MAX_TIMEOUT);} catch (InterruptedException e) {// Ignore, shouldn't be interrupted} finally {cachedConnectionProvider.close(true);if (dialect != null) {dialect.close();}}}
}
  • 详细解释：
    • start 方法：初始化连接器配置，获取数据库连接，启动表监控线程（如果没有自定义查询）。
    • taskClass 方法：返回任务类 JdbcSourceTask。
    • taskConfigs 方法：根据配置生成任务配置。如果有自定义查询，生成一个任务配置；否则，根据表的数量和最大任务数分配表到任务。
    • stop 方法：停止表监控线程，关闭数据库连接和方言。
2. JdbcSourceTask
  • 功能：Kafka Connect 的源任务实现类，负责从 JDBC 数据库中读取数据并生成 Kafka Connect 记录。
  • 关键代码分析：
public class JdbcSourceTask extends SourceTask {// ... 省略部分代码@Overridepublic void start(Map<String, String> properties) {log.info("Starting JDBC source task");try {config = new JdbcSourceTaskConfig(properties);} catch (ConfigException e) {throw new ConfigException("Couldn't start JdbcSourceTask due to configuration error", e);}List<String> tables = config.getList(JdbcSourceTaskConfig.TABLES_CONFIG);Boolean tablesFetched = config.getBoolean(JdbcSourceTaskConfig.TABLES_FETCHED);String query = config.getString(JdbcSourceTaskConfig.QUERY_CONFIG);if ((tables.isEmpty() && query.isEmpty())) {if (!tablesFetched) {// 等待表信息获取taskThreadId.set(Thread.currentThread().getId());log.info("Started JDBC source task. Waiting for DB tables to be fetched.");return;}// 没有分配表或查询throw new ConfigException("Task is being killed because it was not assigned a table nor a query to execute.");}if ((!tables.isEmpty() && !query.isEmpty())) {// 表和查询不能同时分配throw new ConfigException("Invalid configuration: a JdbcSourceTask cannot have both a table and a query assigned to it");}// 获取数据库连接final String url = config.getString(JdbcSourceConnectorConfig.CONNECTION_URL_CONFIG);final int maxConnAttempts = config.getInt(JdbcSourceConnectorConfig.CONNECTION_ATTEMPTS_CONFIG);final long retryBackoff = config.getLong(JdbcSourceConnectorConfig.CONNECTION_BACKOFF_CONFIG);final String dialectName = config.getString(JdbcSourceConnectorConfig.DIALECT_NAME_CONFIG);if (dialectName != null && !dialectName.trim().isEmpty()) {dialect = DatabaseDialects.create(dialectName, config);} else {dialect = DatabaseDialects.findBestFor(url, config);}cachedConnectionProvider = connectionProvider(maxConnAttempts, retryBackoff);// 设置事务隔离级别dialect.setConnectionIsolationMode(cachedConnectionProvider.getConnection(),TransactionIsolationMode.valueOf(config.getString(JdbcSourceConnectorConfig.TRANSACTION_ISOLATION_MODE_CONFIG)));// 确定查询模式TableQuerier.QueryMode queryMode = !query.isEmpty() ? TableQuerier.QueryMode.QUERY : TableQuerier.QueryMode.TABLE;List<String> tablesOrQuery = queryMode == TableQuerier.QueryMode.QUERY ? Collections.singletonList(query) : tables;// 获取偏移量String mode = config.getString(JdbcSourceTaskConfig.MODE_CONFIG);Map<String, List<Map<String, String>>> partitionsByTableFqn = new HashMap<>();Map<Map<String, String>, Map<String, Object>> offsets = null;if (mode.equals(JdbcSourceTaskConfig.MODE_INCREMENTING)|| mode.equals(JdbcSourceTaskConfig.MODE_TIMESTAMP)|| mode.equals(JdbcSourceTaskConfig.MODE_TIMESTAMP_INCREMENTING)) {List<Map<String, String>> partitions = new ArrayList<>(tables.size());switch (queryMode) {case TABLE:for (String table : tables) {List<Map<String, String>> tablePartitions = possibleTablePartitions(table);partitions.addAll(tablePartitions);partitionsByTableFqn.put(table, tablePartitions);}break;case QUERY:partitions.add(Collections.singletonMap(JdbcSourceConnectorConstants.QUERY_NAME_KEY, JdbcSourceConnectorConstants.QUERY_NAME_VALUE));break;default:throw new ConfigException("Unknown query mode: " + queryMode);}offsets = context.offsetStorageReader().offsets(partitions);}// 创建表查询器for (String tableOrQuery : tablesOrQuery) {final List<Map<String, String>> tablePartitionsToCheck;final Map<String, String> partition;switch (queryMode) {case TABLE:tablePartitionsToCheck = partitionsByTableFqn.get(tableOrQuery);break;case QUERY:partition = Collections.singletonMap(JdbcSourceConnectorConstants.QUERY_NAME_KEY, JdbcSourceConnectorConstants.QUERY_NAME_VALUE);tablePartitionsToCheck = Collections.singletonList(partition);break;default:throw new ConfigException("Unexpected query mode: " + queryMode);}Map<String, Object> offset = null;if (offsets != null) {for (Map<String, String> toCheckPartition : tablePartitionsToCheck) {offset = offsets.get(toCheckPartition);if (offset != null) {break;}}}TableQuerier querier = new TableQuerier(queryMode,tableOrQuery,config.getString(JdbcSourceTaskConfig.MODE_CONFIG),config.getString(JdbcSourceTaskConfig.INCREMENTING_COLUMN_NAME_CONFIG),config.getList(JdbcSourceTaskConfig.TIMESTAMP_COLUMN_NAME_CONFIG),config.getLong(JdbcSourceTaskConfig.TIMESTAMP_DELAY_INTERVAL_MS_CONFIG),config.getBoolean(JdbcSourceTaskConfig.VALIDATE_NON_NULL_CONFIG),config.timeZone(),config.getString(JdbcSourceTaskConfig.QUERY_SUFFIX_CONFIG).trim(),offset);tableQueue.add(querier);}}@Overridepublic List<SourceRecord> poll() throws InterruptedException {List<SourceRecord> records = new ArrayList<>();int consecutiveEmptyResults = 0;while (running.get() && records.isEmpty() && consecutiveEmptyResults < CONSECUTIVE_EMPTY_RESULTS_BEFORE_RETURN) {TableQuerier querier = tableQueue.poll();if (querier == null) {consecutiveEmptyResults++;time.sleep(100);continue;}try {List<SourceRecord> querierRecords = querier.poll(cachedConnectionProvider.getConnection(), time);records.addAll(querierRecords);if (querierRecords.isEmpty()) {consecutiveEmptyResults++;} else {consecutiveEmptyResults = 0;}} catch (SQLException e) {log.error("Error polling for table {}: {}", querier.getTableOrQuery(), e.getMessage());maxRetriesPerQuerier--;if (maxRetriesPerQuerier > 0) {time.sleep(1000);tableQueue.add(querier);} else {log.error("Max retries exceeded for table {}", querier.getTableOrQuery());}} finally {if (running.get()) {tableQueue.add(querier);}}}return records;}@Overridepublic void stop() {running.set(false);cachedConnectionProvider.close(true);if (dialect != null) {dialect.close();}}
}
  • 详细解释：
    • start 方法：初始化任务配置，检查配置的有效性，获取数据库连接，设置事务隔离级别，确定查询模式，获取偏移量，创建表查询器。
    • poll 方法：从表查询器队列中取出查询器，执行查询，将查询结果转换为 SourceRecord 列表返回。如果查询失败，进行重试。
    • stop 方法：停止任务，关闭数据库连接和方言。
3. JdbcSourceConnectorConfig
  • 功能：配置类，用于管理 JdbcSourceConnector 的配置信息。
  • 关键代码分析：
public class JdbcSourceConnectorConfig extends AbstractConfig {// ... 省略部分代码public static final String CONNECTION_URL_CONFIG = CONNECTION_PREFIX + "url";public static final String CONNECTION_USER_CONFIG = CONNECTION_PREFIX + "user";public static final String CONNECTION_PASSWORD_CONFIG = CONNECTION_PREFIX + "password";public static final String CONNECTION_ATTEMPTS_CONFIG = CONNECTION_PREFIX + "attempts";public static final String CONNECTION_BACKOFF_CONFIG = CONNECTION_PREFIX + "backoff.ms";public static final String POLL_INTERVAL_MS_CONFIG = "poll.interval.ms";public static final String BATCH_MAX_ROWS_CONFIG = "batch.max.rows";public static final String NUMERIC_PRECISION_MAPPING_CONFIG = "numeric.precision.mapping";public static final String NUMERIC_MAPPING_CONFIG = "numeric.mapping";public static final String DIALECT_NAME_CONFIG = "dialect.name";public static final String MODE_CONFIG = "mode";public static final String INCREMENTING_COLUMN_NAME_CONFIG = "incrementing.column.name";public static final String TIMESTAMP_COLUMN_NAME_CONFIG = "timestamp.column.name";public static final String TIMESTAMP_DELAY_INTERVAL_MS_CONFIG = "timestamp.delay.interval.ms";public static final String VALIDATE_NON_NULL_CONFIG = "validate.non.null";public static final String QUERY_CONFIG = "query";public static final String TABLE_WHITELIST_CONFIG = "table.whitelist";public static final String TABLE_BLACKLIST_CONFIG = "table.blacklist";public static final String TABLE_POLL_INTERVAL_MS_CONFIG = "table.poll.interval.ms";public static final String TABLE_MONITORING_STARTUP_POLLING_LIMIT_MS_CONFIG = "table.monitoring.startup.polling.limit.ms";public static final String TOPIC_PREFIX_CONFIG = "topic.prefix";public static final String TRANSACTION_ISOLATION_MODE_CONFIG = "transaction.isolation.mode";public JdbcSourceConnectorConfig(Map<?, ?> originals) {super(configDef(), originals);}public static ConfigDef configDef() {ConfigDef configDef = new ConfigDef();configDef.define(CONNECTION_URL_CONFIG,Type.STRING,CONNECTION_URL_DEFAULT,Importance.HIGH,CONNECTION_URL_DOC,null,-1,Width.LONG,CONNECTION_URL_DISPLAY);configDef.define(CONNECTION_USER_CONFIG,Type.STRING,null,Importance.MEDIUM,CONNECTION_USER_DOC,null,-1,Width.MEDIUM,CONNECTION_USER_DISPLAY);// ... 省略其他配置项的定义return configDef;}
}
  • 详细解释：
    • 定义了一系列配置项，如数据库连接 URL、用户名、密码、轮询间隔、批量最大行数等。
    • configDef 方法用于定义配置项的元信息，包括类型、默认值、重要性、文档等。

工作流程

  1. 启动连接器：JdbcSourceConnector 的 start 方法被调用，初始化配置，获取数据库连接，启动表监控线程。
  2. 生成任务配置：JdbcSourceConnector 的 taskConfigs 方法根据配置生成任务配置，将表分配到不同的任务中。
  3. 启动任务：JdbcSourceTask 的 start 方法被调用，初始化任务配置，获取数据库连接，设置事务隔离级别，创建表查询器。
  4. 轮询数据：JdbcSourceTask 的 poll 方法被周期性调用，从表查询器队列中取出查询器，执行查询，将查询结果转换为 SourceRecord 列表返回。
  5. 停止任务和连接器：JdbcSourceTask 的 stop 方法和 JdbcSourceConnector 的 stop 方法被调用，关闭数据库连接和方言。

总结

kafka-connect-jdbc 通过 JdbcSourceConnector 和 JdbcSourceTask 实现了从 JDBC 数据库到 Kafka 的数据摄取。JdbcSourceConnector 负责管理连接器的生命周期和任务分配，JdbcSourceTask 负责实际的数据读取和转换。JdbcSourceConnectorConfig 用于管理连接器的配置信息。通过深入理解这些核心类和组件的工作原理，可以更好地使用和扩展 kafka-connect-jdbc。

版权声明:

本网仅为发布的内容提供存储空间，不对发表、转载的内容提供任何形式的保证。凡本网注明“来源：XXX网络”的作品，均转载自其它媒体，著作权归作者所有，商业转载请联系作者获得授权，非商业转载请注明出处。

我们尊重并感谢每一位作者，均已注明文章来源和作者。如因作品内容、版权或其它问题，请及时与我们联系，联系邮箱：809451989@qq.com，投稿邮箱：809451989@qq.com

相关资讯

热文排行

最新新闻

推荐新闻

热搜词