mirror of
https://gitee.com/dromara/dbswitch.git
synced 2026-09-03 05:14:32 +08:00
feat:增加kafka数据库(只支持数据加载写入，不支持变化量同步)
This commit is contained in:
wangjiansong
BIN
dbswitch-admin-ui/src/assets/icons/Kafka.png
Normal file
BIN
dbswitch-admin-ui/src/assets/icons/Kafka.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 18 KiB |
@@ -1 +1 @@
|
||||
<!DOCTYPE html><html><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1"><title>异构数据迁移工具</title><link href=/static/css/app.ab8c389732d782a30e517c00e6a3fcb4.css rel=stylesheet></head><body><div id=app></div><script type=text/javascript src=/static/js/manifest.d06bec2310f88c7766c9.js></script><script type=text/javascript src=/static/js/vendor.8200341f98478c8f7552.js></script><script type=text/javascript src=/static/js/app.94027213e498a69e0bf6.js></script></body></html>
|
||||
<!DOCTYPE html><html><head><meta charset=utf-8><meta name=viewport content="width=device-width,initial-scale=1"><title>异构数据迁移工具</title><link href=/static/css/app.ab8c389732d782a30e517c00e6a3fcb4.css rel=stylesheet></head><body><div id=app></div><script type=text/javascript src=/static/js/manifest.f7865f1f37621bebd5d4.js></script><script type=text/javascript src=/static/js/vendor.8200341f98478c8f7552.js></script><script type=text/javascript src=/static/js/app.94027213e498a69e0bf6.js></script></body></html>
|
||||
BIN
dbswitch-admin/src/main/resources/static/img/Kafka.adb100b.png
Normal file
BIN
dbswitch-admin/src/main/resources/static/img/Kafka.adb100b.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 18 KiB |
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -1,2 +0,0 @@
|
||||
!function(e){var c=window.webpackJsonp;window.webpackJsonp=function(n,a,o){for(var f,d,i,u=0,s=[];u<n.length;u++)d=n[u],r[d]&&s.push(r[d][0]),r[d]=0;for(f in a)Object.prototype.hasOwnProperty.call(a,f)&&(e[f]=a[f]);for(c&&c(n,a,o);s.length;)s.shift()();if(o)for(u=0;u<o.length;u++)i=t(t.s=o[u]);return i};var n={},r={25:0};function t(c){if(n[c])return n[c].exports;var r=n[c]={i:c,l:!1,exports:{}};return e[c].call(r.exports,r,r.exports,t),r.l=!0,r.exports}t.e=function(e){var c=r[e];if(0===c)return new Promise(function(e){e()});if(c)return c[2];var n=new Promise(function(n,t){c=r[e]=[n,t]});c[2]=n;var a=document.getElementsByTagName("head")[0],o=document.createElement("script");o.type="text/javascript",o.charset="utf-8",o.async=!0,o.timeout=12e4,t.nc&&o.setAttribute("nonce",t.nc),o.src=t.p+"static/js/"+e+"."+{0:"13280e0c66c3dac0ad66",1:"fed755dff73c923d7f16",2:"22e821ec8c9909429f09",3:"fe1c5c7d88360f5f98ae",4:"08d31a8e3923090c336f",5:"2ba81da755c2838a1bc0",6:"5d5ddd62bcebfdd45c40",7:"5832297a8db38b5a6b9a",8:"fcf46f880cde60008573",9:"7482dae9c15b18627401",10:"80e7801109b57f1a3d72",11:"1b5e1420657152e07dc2",12:"ddd8e2a887b327f4ab2d",13:"a1fc0f5e975ae62c9069",14:"572f60f49a9868fcd0fe",15:"78a79600adcd06361937",16:"66f213fc135ae6a5d988",17:"bfc722f22acf2bd97223",18:"f7e4b41780ec8b255f7a",19:"852c4c5cdec46703f5a8",20:"a476a38191cde522905a",21:"e2737863c3879ad32e68",22:"ac430228b1464a487c04"}[e]+".js";var f=setTimeout(d,12e4);function d(){o.onerror=o.onload=null,clearTimeout(f);var c=r[e];0!==c&&(c&&c[1](new Error("Loading chunk "+e+" failed.")),r[e]=void 0)}return o.onerror=o.onload=d,a.appendChild(o),n},t.m=e,t.c=n,t.d=function(e,c,n){t.o(e,c)||Object.defineProperty(e,c,{configurable:!1,enumerable:!0,get:n})},t.n=function(e){var c=e&&e.__esModule?function(){return e.default}:function(){return e};return t.d(c,"a",c),c},t.o=function(e,c){return Object.prototype.hasOwnProperty.call(e,c)},t.p="/",t.oe=function(e){throw console.error(e),e}}([]);
|
||||
//# sourceMappingURL=manifest.d06bec2310f88c7766c9.js.map
|
||||
@@ -0,0 +1,2 @@
|
||||
!function(e){var n=window.webpackJsonp;window.webpackJsonp=function(r,a,o){for(var f,d,i,u=0,b=[];u<r.length;u++)d=r[u],t[d]&&b.push(t[d][0]),t[d]=0;for(f in a)Object.prototype.hasOwnProperty.call(a,f)&&(e[f]=a[f]);for(n&&n(r,a,o);b.length;)b.shift()();if(o)for(u=0;u<o.length;u++)i=c(c.s=o[u]);return i};var r={},t={25:0};function c(n){if(r[n])return r[n].exports;var t=r[n]={i:n,l:!1,exports:{}};return e[n].call(t.exports,t,t.exports,c),t.l=!0,t.exports}c.e=function(e){var n=t[e];if(0===n)return new Promise(function(e){e()});if(n)return n[2];var r=new Promise(function(r,c){n=t[e]=[r,c]});n[2]=r;var a=document.getElementsByTagName("head")[0],o=document.createElement("script");o.type="text/javascript",o.charset="utf-8",o.async=!0,o.timeout=12e4,c.nc&&o.setAttribute("nonce",c.nc),o.src=c.p+"static/js/"+e+"."+{0:"01b50a1b04adf9f0bb05",1:"fed755dff73c923d7f16",2:"22e821ec8c9909429f09",3:"fe1c5c7d88360f5f98ae",4:"08d31a8e3923090c336f",5:"2ba81da755c2838a1bc0",6:"5d5ddd62bcebfdd45c40",7:"5832297a8db38b5a6b9a",8:"fcf46f880cde60008573",9:"7482dae9c15b18627401",10:"80e7801109b57f1a3d72",11:"1b5e1420657152e07dc2",12:"ddd8e2a887b327f4ab2d",13:"a1fc0f5e975ae62c9069",14:"572f60f49a9868fcd0fe",15:"78a79600adcd06361937",16:"66f213fc135ae6a5d988",17:"bfc722f22acf2bd97223",18:"f7e4b41780ec8b255f7a",19:"852c4c5cdec46703f5a8",20:"a476a38191cde522905a",21:"e2737863c3879ad32e68",22:"ac430228b1464a487c04"}[e]+".js";var f=setTimeout(d,12e4);function d(){o.onerror=o.onload=null,clearTimeout(f);var n=t[e];0!==n&&(n&&n[1](new Error("Loading chunk "+e+" failed.")),t[e]=void 0)}return o.onerror=o.onload=d,a.appendChild(o),r},c.m=e,c.c=r,c.d=function(e,n,r){c.o(e,n)||Object.defineProperty(e,n,{configurable:!1,enumerable:!0,get:r})},c.n=function(e){var n=e&&e.__esModule?function(){return e.default}:function(){return e};return c.d(n,"a",n),n},c.o=function(e,n){return Object.prototype.hasOwnProperty.call(e,n)},c.p="/",c.oe=function(e){throw console.error(e),e}}([]);
|
||||
//# sourceMappingURL=manifest.f7865f1f37621bebd5d4.js.map
|
||||
File diff suppressed because one or more lines are too long
@@ -241,6 +241,15 @@ public enum ProductTypeEnum {
|
||||
"jdbc:uxdb://",
|
||||
new String[]{"jdbc:uxdb://{host}[:{port}]/[{database}][\\?{params}]"},
|
||||
"jdbc:uxdb://127.0.0.1:52025/uxdb"),
|
||||
/**
|
||||
* KAFKA数据库类型
|
||||
*/
|
||||
KAFKA(25, "\"", "Kafka", "org.apache.kafka.connect.jdbc.JdbcSinkConnector", 9092,
|
||||
"SELECT 1",
|
||||
"jdbc:kafka://",
|
||||
new String[]{"jdbc:kafka://{host}[:{port}]/[{database}][\\?{params}]"},
|
||||
"jdbc:kafka://127.0.0.1:9092/test"),
|
||||
|
||||
|
||||
;
|
||||
|
||||
|
||||
28
dbswitch-product/dbswitch-product-kafka/pom.xml
Normal file
28
dbswitch-product/dbswitch-product-kafka/pom.xml
Normal file
@@ -0,0 +1,28 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<groupId>org.dromara.dbswitch</groupId>
|
||||
<artifactId>dbswitch-product</artifactId>
|
||||
<version>2.0.1</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>dbswitch-product-kafka</artifactId>
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.dromara.dbswitch</groupId>
|
||||
<artifactId>dbswitch-common</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.dromara.dbswitch</groupId>
|
||||
<artifactId>dbswitch-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,67 @@
|
||||
// Copyright tang. All rights reserved.
|
||||
// https://gitee.com/inrgihc/dbswitch
|
||||
//
|
||||
// Use of this source code is governed by a BSD-style license
|
||||
//
|
||||
// Author: tang (inrgihc@126.com)
|
||||
// Date : 2020/1/2
|
||||
// Location: beijing , china
|
||||
/////////////////////////////////////////////////////////////
|
||||
package org.dromara.dbswitch.product.kafka;
|
||||
|
||||
import org.dromara.dbswitch.core.annotation.Product;
|
||||
import org.dromara.dbswitch.common.type.ProductTypeEnum;
|
||||
import org.dromara.dbswitch.core.features.DefaultProductFeatures;
|
||||
import org.dromara.dbswitch.core.features.ProductFeatures;
|
||||
import org.dromara.dbswitch.core.provider.AbstractFactoryProvider;
|
||||
import org.dromara.dbswitch.core.provider.meta.MetadataProvider;
|
||||
import org.dromara.dbswitch.core.provider.manage.TableManageProvider;
|
||||
import org.dromara.dbswitch.core.provider.query.TableDataQueryProvider;
|
||||
import org.dromara.dbswitch.core.provider.sync.TableDataSynchronizeProvider;
|
||||
import org.dromara.dbswitch.core.provider.write.TableDataWriteProvider;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
/**
|
||||
* Kafka工厂提供者
|
||||
* Kafka作为消息队列系统，不支持传统的关系型数据库操作
|
||||
* 主要用于数据写入场景
|
||||
*/
|
||||
@Product(ProductTypeEnum.KAFKA)
|
||||
public class KafkaFactoryProvider extends AbstractFactoryProvider {
|
||||
|
||||
public KafkaFactoryProvider(DataSource dataSource) {
|
||||
super(dataSource);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ProductFeatures getProductFeatures() {
|
||||
return new DefaultProductFeatures();
|
||||
}
|
||||
|
||||
@Override
|
||||
public MetadataProvider createMetadataQueryProvider() {
|
||||
return new KafkaMetadataQueryProvider(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableDataQueryProvider createTableDataQueryProvider() {
|
||||
// Kafka不支持数据查询，返回null或使用默认实现
|
||||
return super.createTableDataQueryProvider();
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableDataWriteProvider createTableDataWriteProvider(boolean useInsert) {
|
||||
return new KafkaTableDataWriteProvider(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableManageProvider createTableManageProvider() {
|
||||
return new KafkaTableManageProvider(this);
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableDataSynchronizeProvider createTableDataSynchronizeProvider() {
|
||||
return new KafkaTableDataSynchronizer(this);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
// Copyright tang. All rights reserved.
|
||||
// https://gitee.com/inrgihc/dbswitch
|
||||
//
|
||||
// Use of this source code is governed by a BSD-style license
|
||||
//
|
||||
// Author: tang (inrgihc@126.com)
|
||||
// Date : 2020/1/2
|
||||
// Location: beijing , china
|
||||
/////////////////////////////////////////////////////////////
|
||||
package org.dromara.dbswitch.product.kafka;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
|
||||
import org.dromara.dbswitch.core.provider.meta.AbstractMetadataProvider;
|
||||
import org.dromara.dbswitch.core.schema.ColumnDescription;
|
||||
import org.dromara.dbswitch.core.schema.ColumnMetaData;
|
||||
import org.dromara.dbswitch.core.schema.SourceProperties;
|
||||
import org.dromara.dbswitch.core.schema.TableDescription;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Kafka元数据查询提供者
|
||||
* Kafka作为消息队列，其"表"概念对应Topic
|
||||
*/
|
||||
@Slf4j
|
||||
public class KafkaMetadataQueryProvider extends AbstractMetadataProvider {
|
||||
|
||||
public KafkaMetadataQueryProvider(ProductFactoryProvider factoryProvider) {
|
||||
super(factoryProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> querySchemaList(Connection connection) {
|
||||
// Kafka没有schema概念，返回空列表或默认值
|
||||
List<String> schemas = new ArrayList<>();
|
||||
schemas.add("default");
|
||||
return schemas;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<TableDescription> queryTableList(Connection connection, String schemaName) {
|
||||
// Kafka的"表"对应Topic，这里需要通过JDBC驱动获取Topic列表
|
||||
// 由于Kafka JDBC驱动的特殊性，返回空列表或需要从配置中获取
|
||||
log.warn("Kafka does not support traditional table listing. Topics should be configured externally.");
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public TableDescription queryTableMeta(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka的"表"对应Topic
|
||||
TableDescription td = new TableDescription();
|
||||
td.setSchemaName(schemaName);
|
||||
td.setTableName(tableName);
|
||||
td.setTableType("TOPIC");
|
||||
td.setRemarks("Kafka Topic: " + tableName);
|
||||
return td;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> queryTableColumnName(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka消息没有固定的列结构，返回空列表
|
||||
log.warn("Kafka topics do not have fixed column structures.");
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ColumnDescription> queryTableColumnMeta(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka消息没有固定的列结构
|
||||
log.warn("Kafka topics do not have fixed column metadata.");
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> queryTablePrimaryKeys(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka不支持主键概念
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getTableDDL(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka不支持传统DDL
|
||||
return "-- Kafka Topic: " + tableName + "\n-- DDL not applicable for Kafka";
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getViewDDL(Connection connection, String schemaName, String tableName) {
|
||||
// Kafka不支持视图
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ColumnDescription> querySelectSqlColumnMeta(Connection connection, String sql) {
|
||||
// Kafka不支持SQL查询
|
||||
log.warn("Kafka does not support SQL queries for metadata extraction.");
|
||||
return new ArrayList<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void testQuerySQL(Connection connection, String sql) {
|
||||
// Kafka不支持SQL查询测试
|
||||
log.warn("Kafka does not support SQL query testing.");
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getFieldDefinition(ColumnMetaData v, List<String> pks, boolean useAutoInc,
|
||||
boolean addCr, boolean withRemarks) {
|
||||
// Kafka不需要字段定义
|
||||
return "-- Field definition not applicable for Kafka";
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> getTableColumnCommentDefinition(TableDescription td, List<ColumnDescription> cds) {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void preAppendCreateTableSql(StringBuilder builder) {
|
||||
// Kafka不需要建表前缀
|
||||
}
|
||||
|
||||
@Override
|
||||
public void postAppendCreateTableSql(StringBuilder builder, String tblComment, List<String> primaryKeys,
|
||||
SourceProperties tblProperties) {
|
||||
// Kafka不需要建表后缀
|
||||
builder.append("-- Kafka topic creation is managed externally");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
// Copyright tang. All rights reserved.
|
||||
// https://gitee.com/inrgihc/dbswitch
|
||||
//
|
||||
// Use of this source code is governed by a BSD-style license
|
||||
//
|
||||
// Author: tang (inrgihc@126.com)
|
||||
// Date : 2020/1/2
|
||||
// Location: beijing , china
|
||||
/////////////////////////////////////////////////////////////
|
||||
package org.dromara.dbswitch.product.kafka;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
|
||||
import org.dromara.dbswitch.core.provider.sync.DefaultTableDataSynchronizeProvider;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Kafka表数据同步提供者
|
||||
* Kafka作为消息队列，主要支持数据插入操作（发送消息）
|
||||
* 不支持传统的update和delete操作
|
||||
*/
|
||||
@Slf4j
|
||||
public class KafkaTableDataSynchronizer extends DefaultTableDataSynchronizeProvider {
|
||||
|
||||
public KafkaTableDataSynchronizer(ProductFactoryProvider factoryProvider) {
|
||||
super(factoryProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void prepare(String schemaName, String tableName, List<String> fieldNames, List<String> pks) {
|
||||
// Kafka不需要主键概念，但为了兼容性仍然调用父类方法
|
||||
log.info("Preparing Kafka data synchronizer for topic: {}", tableName);
|
||||
|
||||
// 初始化基本字段，但不依赖传统数据库的元数据
|
||||
this.columnType = Collections.emptyMap();
|
||||
this.fieldOrders = fieldNames;
|
||||
this.pksOrders = pks != null ? pks : Collections.emptyList();
|
||||
|
||||
// 为Kafka构建INSERT语句模板
|
||||
this.insertStatementSql = getInsertPrepareStatementSql(schemaName, tableName, fieldNames);
|
||||
|
||||
// Kafka不支持UPDATE和DELETE，设置为null或空
|
||||
this.updateStatementSql = null;
|
||||
this.deleteStatementSql = null;
|
||||
|
||||
// 设置参数类型数组
|
||||
this.insertArgsType = new int[fieldNames.size()];
|
||||
// 对于Kafka，所有字段都视为字符串类型
|
||||
for (int i = 0; i < fieldNames.size(); i++) {
|
||||
insertArgsType[i] = java.sql.Types.VARCHAR;
|
||||
}
|
||||
|
||||
this.updateArgsType = new int[0];
|
||||
this.deleteArgsType = new int[0];
|
||||
}
|
||||
|
||||
@Override
|
||||
public long executeInsert(List<Object[]> records) {
|
||||
if (records == null || records.isEmpty()) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
log.info("Executing insert of {} records to Kafka topic", records.size());
|
||||
|
||||
// 使用父类的插入逻辑，通过JDBC驱动向Kafka发送消息
|
||||
try {
|
||||
return super.executeInsert(records);
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to insert records to Kafka", e);
|
||||
throw new RuntimeException("Failed to send messages to Kafka topic", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long executeUpdate(List<Object[]> records) {
|
||||
// Kafka不支持更新操作
|
||||
log.warn("Kafka does not support update operations. Records will be ignored.");
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long executeDelete(List<Object[]> records) {
|
||||
// Kafka不支持删除操作
|
||||
log.warn("Kafka does not support delete operations. Records will be ignored.");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
// Copyright tang. All rights reserved.
|
||||
// https://gitee.com/inrgihc/dbswitch
|
||||
//
|
||||
// Use of this source code is governed by a BSD-style license
|
||||
//
|
||||
// Author: tang (inrgihc@126.com)
|
||||
// Date : 2020/1/2
|
||||
// Location: beijing , china
|
||||
/////////////////////////////////////////////////////////////
|
||||
package org.dromara.dbswitch.product.kafka;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.commons.collections4.CollectionUtils;
|
||||
import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
|
||||
import org.dromara.dbswitch.core.provider.write.DefaultTableDataWriteProvider;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.sql.SQLException;
|
||||
import java.sql.Statement;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Kafka表数据写入提供者
|
||||
* Kafka作为消息队列，数据写入对应于向Topic发送消息
|
||||
*/
|
||||
@Slf4j
|
||||
public class KafkaTableDataWriteProvider extends DefaultTableDataWriteProvider {
|
||||
|
||||
public KafkaTableDataWriteProvider(ProductFactoryProvider factoryProvider) {
|
||||
super(factoryProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void prepareWrite(String schemaName, String tableName, List<String> fieldNames) {
|
||||
// Kafka不需要预处理，但需要设置schema和tableName
|
||||
this.columnType = Collections.emptyMap();
|
||||
this.schemaName = schemaName;
|
||||
this.tableName = tableName;
|
||||
|
||||
log.info("Preparing to write to Kafka topic: {}", tableName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public long write(List<String> fieldNames, List<Object[]> recordValues) {
|
||||
if (CollectionUtils.isEmpty(fieldNames) || CollectionUtils.isEmpty(recordValues)) {
|
||||
return 0L;
|
||||
}
|
||||
|
||||
log.info("Writing {} records to Kafka topic: {}", recordValues.size(), tableName);
|
||||
|
||||
// 对于Kafka，我们需要将记录转换为消息并发送到Topic
|
||||
// 这里使用JDBC驱动提供的机制来发送数据
|
||||
try (Connection connection = getDataSource().getConnection()) {
|
||||
for (Object[] record : recordValues) {
|
||||
// 构建INSERT语句用于Kafka JDBC连接器
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append("INSERT INTO ").append(tableName).append(" (");
|
||||
|
||||
for (int i = 0; i < fieldNames.size(); i++) {
|
||||
if (i > 0) sb.append(", ");
|
||||
sb.append(fieldNames.get(i));
|
||||
}
|
||||
sb.append(") VALUES (");
|
||||
|
||||
for (int i = 0; i < record.length; i++) {
|
||||
if (i > 0) sb.append(", ");
|
||||
if (record[i] == null) {
|
||||
sb.append("NULL");
|
||||
} else {
|
||||
sb.append("'").append(record[i].toString().replace("'", "''")).append("'");
|
||||
}
|
||||
}
|
||||
sb.append(")");
|
||||
|
||||
try (Statement stmt = connection.createStatement()) {
|
||||
stmt.executeUpdate(sb.toString());
|
||||
}
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new RuntimeException("Failed to write data to Kafka topic: " + tableName, e);
|
||||
}
|
||||
|
||||
return recordValues.size();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
// Copyright tang. All rights reserved.
|
||||
// https://gitee.com/inrgihc/dbswitch
|
||||
//
|
||||
// Use of this source code is governed by a BSD-style license
|
||||
//
|
||||
// Author: tang (inrgihc@126.com)
|
||||
// Date : 2020/1/2
|
||||
// Location: beijing , china
|
||||
/////////////////////////////////////////////////////////////
|
||||
package org.dromara.dbswitch.product.kafka;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.dbswitch.core.provider.ProductFactoryProvider;
|
||||
import org.dromara.dbswitch.core.provider.manage.DefaultTableManageProvider;
|
||||
|
||||
/**
|
||||
* Kafka表管理提供者
|
||||
* Kafka作为消息队列，不支持传统的表操作如truncate和drop
|
||||
*/
|
||||
@Slf4j
|
||||
public class KafkaTableManageProvider extends DefaultTableManageProvider {
|
||||
|
||||
public KafkaTableManageProvider(ProductFactoryProvider factoryProvider) {
|
||||
super(factoryProvider);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void truncateTableData(String schemaName, String tableName) {
|
||||
// Kafka不支持truncate操作，Topic中的数据通过保留策略自动清理
|
||||
log.warn("Kafka does not support truncate operation. Topic data is managed by retention policy.");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void dropTable(String schemaName, String tableName) {
|
||||
// Kafka不支持drop操作，Topic需要通过Kafka管理工具删除
|
||||
log.warn("Kafka does not support drop operation. Topics should be managed through Kafka admin tools.");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
org.dromara.dbswitch.product.kafka.KafkaFactoryProvider
|
||||
@@ -133,6 +133,11 @@
|
||||
<artifactId>dbswitch-product-tdengine</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.dromara.dbswitch</groupId>
|
||||
<artifactId>dbswitch-product-kafka</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
||||
@@ -37,6 +37,7 @@
|
||||
<module>dbswitch-product-doris</module>
|
||||
<module>dbswitch-product-oceanbase</module>
|
||||
<module>dbswitch-product-tdengine</module>
|
||||
<module>dbswitch-product-kafka</module>
|
||||
</modules>
|
||||
|
||||
</project>
|
||||
|
||||
@@ -179,3 +179,10 @@ jdbc驱动名称：com.gitee.jdbc.elasticsearch.JdbcDriver
|
||||
```
|
||||
|
||||
> 项目地址：https://gitee.com/inrgihc/jdbc-jest-driver
|
||||
|
||||
**Kafka数据库**
|
||||
|
||||
```
|
||||
jdbc连接地址：jdbc:kafka://127.0.0.1:9092/test
|
||||
jdbc驱动名称：org.apache.kafka.connect.jdbc.JdbcSinkConnector
|
||||
```
|
||||
|
||||
BIN
drivers/kafka/kafka-clients-0.11.0.0/kafka-clients-0.11.0.0.jar
Normal file
BIN
drivers/kafka/kafka-clients-0.11.0.0/kafka-clients-0.11.0.0.jar
Normal file
Binary file not shown.
Reference in New Issue
Block a user