mydataharbor

mydataharbor-elasticsearch

mydataharbor-http

mydataharbor-jdbc

mydataharbor-kafka

mydataharbor-redis

2026年04月03日 Release Downloads 插件列表

📖 简介

MyDataHarbor 是一个分布式、高扩展、高性能的准实时数据同步中间件,致力于解决异构数据源之间的数据同步问题。

🎯 核心定位:为实时交易系统服务,支持海量数据的准实时增量同步和定时全量同步,亦可用于 ETL 领域。

典型应用场景:

• 🔄 数据库 → 搜索引擎:MySQL → Elasticsearch
• 🔄 数据库 → 缓存:MySQL → Redis
• 🔄 消息队列 → 数据库:Kafka → MySQL
• 🔄 跨部门数据同步:多系统间的数据共享与同步

✨ 核心特性

🏗️ 架构设计

MyDataHarbor 唯一依赖的中间件是 ZooKeeper,共有两个核心组件:

集群架构设计

集群设计

节点任务设计

节点任务设计

系统架构(示意)

 ┌─────────────────────────────────────────┐
 │ MyDataHarbor Console │
 │ - 任务管理 - 插件管理 - 监控看板 │
 └───────────────────┬─────────────────────┘
 │
 ┌─────────▼─────────┐
 │ ZooKeeper │
 │ (配置中心/协调) │
 └─────────┬─────────┘
 │
 ┌─────────────────────────────┼─────────────────────────────┐
 │ │ │
 ┌───────▼───────┐ ┌───────▼───────┐ ┌───────▼───────┐
 │ Server A │ │ Server B │ │ Server C │
 │ (group: G1) │ │ (group: G1) │ │ (group: G2) │
 │ - Task 1 │ │ - Task 2 │ │ - Task 3 │
 │ - Plugin │ │ - Plugin │ │ - Plugin │
 └───────────────┘ └───────────────┘ └───────────────┘

核心组件

数据处理流程

┌──────────────┐ ┌───────────────┐ ┌───────────┐ ┌───────────────┐ ┌──────────────┐
│ DataSource │ → │ ProtocolData │ → │ Checker │ → │ DataConverter │ → │ DataSink │
│ (数据源) │ │ (协议转换) │ │ (数据校验) │ │ (数据转换) │ │ (数据写入) │
└──────────────┘ └───────────────┘ └───────────┘ └───────────────┘ └──────────────┘
 ↓ ↓ ↓ ↓
 MySQL JSON POJO Elasticsearch
 Kafka XML Map Redis
 RabbitMQ ... ... ...

🚀 快速开始

环境要求

方式一:下载二进制包

1. 下载安装包

从 # Linux wget https://github.com/mydataharbor/mydataharbor/releases/download/v2.0.1/mydataharbor-console-2.0.1-bin.tar.gz wget https://github.com/mydataharbor/mydataharbor/releases/download/v2.0.1/mydataharbor-server-2.0.1-bin.tar.gz # 解压 tar -xzf mydataharbor-console-2.0.1-bin.tar.gz tar -xzf mydataharbor-server-2.0.1-bin.tar.gz

server:
 port: 8080 # Console 服务端口
zk: 127.0.0.1:2181 # ZooKeeper 地址

zk: ["127.0.0.1:2181"] # ZooKeeper 地址
port: 1299 # Server 服务端口
group: biz001 # 节点所属组名
pluginRepository: http://127.0.0.1:8080 # 插件仓库地址(Console)

# 启动 Console(管理台)
cd mydataharbor-console
./start.sh

# 启动 Server(数据同步节点)
cd mydataharbor-server
./start.sh

# 克隆项目
git clone https://github.com/mydataharbor/mydataharbor.git
cd mydataharbor

# 编译打包
mvn clean package -DskipTests

# 产物位置
# mydataharbor-deploy/mydataharbor-console/target/mydataharbor-console-{version}-bin.tar.gz
# mydataharbor-deploy/mydataharbor-server/target/mydataharbor-server-{version}-bin.tar.gz

# 启动 ZooKeeper
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4

# 启动 Console
docker run -d --name mydataharbor-console \
 -e ZK_HOST=127.0.0.1:2181 \
 -p 8080:8080 \
 mydataharbor/mydataharbor-console:2.0.1

# 启动 Server
docker run -d --name mydataharbor-server \
 -e ZK_HOST=127.0.0.1:2181 \
 -e GROUP=biz001 \
 -e PLUGIN_REPO=http://127.0.0.1:8080 \
 -p 1299:1299 \
 mydataharbor/mydataharbor-server:2.0.1

myplugin/
├── pom.xml # Maven 配置
├── plugin.properties # 插件描述文件
└── src/main/java/
 └── myplugin/
 ├── MyDataSource.java # 数据源实现
 ├── MyDataSink.java # 写入器实现
 ├── MyConverter.java # 转换器实现
 └── MyPipelineCreator.java # Pipeline 创建器

# 插件唯一标识
plugin.id=myplugin
# 插件版本
plugin.version=1.0.0
# 插件描述
plugin.desc=我的自定义插件
# 插件作者
plugin.author=yourname
# 依赖其他插件(可选)
plugin.dependencies=mydataharbor-base:1.0.0,mysql-plugin:2.0.0
# 插件依赖类加载器隔离
plugin.requiresIsolation=true

@MyDataHarborMarker(title = "我的数据源")
@Extension
public class MyDataSource implements IDataSource<MyRecord, MySetting> {

 // 数据总数(用于进度展示)
 private long total;

 // 当前进度
 private long now = 0L;

 @Override
 public void init(MySetting setting) {
 // 初始化资源,如数据库连接
 this.total = queryTotalCount();
 }

 @Override
 public String dataSourceType() {
 return "MyDataSource";
 }

 @Override
 public Long total() {
 return total;
 }

 @Override
 public Iterable<MyRecord> poll(MySetting setting) throws TheEndException {
 // 当数据全部拉取完成时抛出 TheEndException
 if (now >= total) {
 throw new TheEndException("数据拉取完成");
 }

 List<MyRecord> batch = new ArrayList<>();
 // 批量拉取数据(建议每批 100-1000 条)
 while (now < total && batch.size() < 100) {
 MyRecord record = fetchData(now);
 batch.add(record);
 now++;
 }
 return batch;
 }

 @Override
 public void commit(MyRecord record, MySetting setting) {
 // 单条提交(记录成功位置,用于断点续传)
 }

 @Override
 public void commit(Iterable<MyRecord> records, MySetting setting) {
 // 批量提交(推荐实现此方法以提升性能)
 for (MyRecord record : records) {
 commit(record, setting);
 }
 }

 @Override
 public void rollback(MyRecord record, MySetting setting) {
 // 单条回滚(处理失败记录)
 }

 @Override
 public void close() throws IOException {
 // 释放资源
 }
}

@MyDataHarborMarker(title = "我的写入器")
@Extension
@Slf4j
public class MyDataSink implements IDataSink<MyRecord, MySetting> {

 @Override
 public String name() {
 return "MyDataSink";
 }

 @Override
 public WriterResult write(MyRecord record, MySetting setting) throws ResetException {
 // 单条写入
 try {
 // 执行写入操作
 doWrite(record);
 return WriterResult.builder()
 .commit(true)
 .success(true)
 .msg("success")
 .build();
 } catch (Exception e) {
 log.error("写入失败", e);
 return WriterResult.builder()
 .commit(false)
 .success(false)
 .msg(e.getMessage())
 .build();
 }
 }

 @Override
 public WriterResult write(List<MyRecord> records, MySetting setting) throws ResetException {
 // 批量写入(推荐实现,性能更高)
 try {
 doBatchWrite(records);
 return WriterResult.builder()
 .commit(true)
 .success(true)
 .msg("batch success")
 .build();
 } catch (Exception e) {
 log.error("批量写入失败", e);
 return WriterResult.builder()
 .commit(false)
 .success(false)
 .msg(e.getMessage())
 .build();
 }
 }

 @Override
 public void close() throws IOException {
 // 释放资源
 }
}

@MyDataHarborMarker(title = "我的转换器")
@Extension
public class MyDataConverter implements IDataConverter<ProtocolData, MyRecord> {

 @Override
 public MyRecord convert(ProtocolData protocolData, BaseSettingContext settingContext) {
 // 将 ProtocolData 转换为业务 Record 对象
 MyRecord record = new MyRecord();
 record.setId(protocolData.getId());
 record.setData(protocolData.getContent());
 return record;
 }
}

@MyDataHarborMarker(title = "我的校验器")
@Extension
public class MyChecker implements IChecker<MyRecord> {

 @Override
 public CheckResult check(MyRecord record, BaseSettingContext settingContext) {
 // 数据校验逻辑
 if (record == null || record.getId() == null) {
 return CheckResult.fail("数据不完整");
 }
 return CheckResult.pass();
 }
}

@Extension
public class MyPipelineCreator extends AbstractAutoScanPipelineCreator<MyConfig, MySetting> {

 @Override
 public String type() {
 return "我的数据同步任务";
 }

 @Override
 public IDataPipeline createPipeline(MyConfig config, MySetting setting) throws Exception {
 return CommonDataPipeline.builder()
 .dataSource(new MyDataSource())
 .protocolDataConverter(new OriginalProtocolDataConverter())
 .checker(new MyChecker())
 .dataConverter(new MyDataConverter())
 .sink(new MyDataSink())
 .settingContext(setting)
 .build();
 }

 @Override
 public String scanPackage() {
 return "myplugin";
 }

 // 配置类(用于生成可视化配置界面)
 @Data
 public static class MyConfig {
 @MyDataHarborMarker(title = "数据源地址", defaultValue = "localhost:3306")
 private String dataSourceUrl;

 @MyDataHarborMarker(title = "用户名", defaultValue = "root")
 private String username;

 @MyDataHarborMarker(title = "密码", defaultValue = "", secret = true)
 private String password;

 @MyDataHarborMarker(title = "批处理大小", defaultValue = "1000")
 private Integer batchSize;
 }
}

# 依赖基础插件(版本 1.0.0 及以上)
plugin.dependencies=mydataharbor-base:1.0.0

# 依赖多个插件
plugin.dependencies=mysql-plugin:2.0.0,elasticsearch-plugin:1.5.0

# 可选依赖(不存在也不影响运行)
plugin.dependencies=mysql-plugin:2.0.0?optional=true

# 编译打包
mvn clean package

# 产物位置
target/myplugin-1.0.0.zip

# 发布到插件仓库
# 方式 1:上传到 Console 管理台
curl -X POST http://localhost:8080/mydataharbor/plugin/upload \
 -F "file=@target/myplugin-1.0.0.zip"

# 方式 2:手动放置到插件仓库目录
cp target/myplugin-1.0.0.zip /path/to/plugin-repository/

# 方式 1:将插件 JAR 放到 Server 插件目录
cp target/myplugin-1.0.0.zip mydataharbor-server/plugins/
# 重启 Server

# 方式 2:Maven 依赖方式开发
# 在 Server 项目的 pom.xml 中添加:
<dependency>
 <groupId>com.mydataharbor</groupId>
 <artifactId>myplugin</artifactId>
 <version>1.0.0</version>
</dependency>

pipelineType: mysql-to-elasticsearch
settingContext:
 dataSource:
 jdbcUrl: jdbc:mysql://localhost:3306/mydb
 username: root
 password: "123456"
 tables: ["users", "orders"]
 sink:
 esHosts: ["localhost:9200"]
 indexPrefix: "myapp"
 batchCommit: true
 batchSize: 1000
 parallel: true
 threadNum: 4

pipelineType: kafka-to-mysql
settingContext:
 kafka:
 bootstrapServers: ["localhost:9092"]
 topic: "user-events"
 groupId: "mysql-sync-group"
 mysql:
 jdbcUrl: jdbc:mysql://localhost:3306/target_db
 username: root
 password: "123456"
 table: "events"
 batchCommit: true
 batchSize: 500

# API 方式
curl http://localhost:8080/mydataharbor/task/list?group=biz001

# JMX 方式
# 连接 JMX 端口 9999,查看 MyDataHarbor MBean

# 暂停任务
curl -X POST http://localhost:8080/mydataharbor/task/pause?taskId=task-001

# 恢复任务
curl -X POST http://localhost:8080/mydataharbor/task/resume?taskId=task-001

server:
 port: 8080 # Web 服务端口
 servlet:
 multipart:
 max-file-size: 500MB # 插件上传大小限制
 max-request-size: 500MB

zk: 127.0.0.1:2181 # ZooKeeper 连接地址

spring:
 devtools:
 livereload:
 enabled: true # 开发环境热加载

zk: ["127.0.0.1:2181"] # ZooKeeper 集群地址
port: 1299 # RPC 服务端口
group: biz001 # 节点分组名(同组节点负载均衡)

pluginRepository: http://127.0.0.1:8080 # 插件仓库地址

# 可选:JMX 监控配置
jmx:
 enabled: true
 port: 9999

./start.sh jmx

# 任务配置
settingContext:
 batchCommit: true # 批量提交
 batchSize: 1000 # 每批 1000 条
 parallel: true # 开启并行处理
 threadNum: 4 # 4 线程并行

# 任务配置
enableRebalance: true # 开启故障转移
enableLoadBalance: true # 开启负载均衡

# 克隆项目
git clone https://github.com/mydataharbor/mydataharbor.git
cd mydataharbor

# 导入 IDE(IntelliJ IDEA 推荐)
# File -> Open -> 选择项目根目录

# 编译
mvn clean install

# 运行单元测试
mvn test

feat: 新增 XXX 功能
fix: 修复 XXX 问题
docs: 文档更新
refactor: 代码重构
test: 测试用例
chore: 构建/配置变更