┌─────────────────────────────────────────┐
                    │           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               ...                                     ...               ...

# 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: 构建/配置变更

Copyright (C) 2020-2026 xulang <[email protected]>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.