benfoster/ditto-kinesis

Ditto-Kinesis 事件流复制工具

镜像概述

Ditto-Kinesis是针对Event Store的集群复制工具，通过订阅源集群特定事件流，将其复制到Kinesis或目标集群。它可作为独立应用通过Docker运行，利用Event Store订阅组实现竞争消费，并集成Serilog日志、Seq（本地）和Datadog（生产）的日志管理。

核心功能

1. 事件流复制：订阅源Event Store集群指定流并复制到目标端；
2. 灵活配置：支持JSON文件（appsettings.json）或环境变量配置；
3. 幂等性保证：基于Event Store流和事件ID提供幂等性，启用版本检查时确保完全幂等；
4. 指标监控：在http://localhost:5000/metrics暴露Prometheus格式指标；
5. 负载控制：通过缓冲区大小和复制节流间隔调整源服务器负载。

使用场景

• 跨环境Event Store集群事件同步；
• 事件数据备份与容灾；
• 构建事件驱动架构的数据流管道；
• 从Event Store向Kinesis传输事件以支持数据分析。

配置说明

主要配置项对应关系：

Docker部署示例

拉取镜像

bash
docker pull benfoster/ditto-kinesis

本地运行（Docker Compose）

克隆仓库后执行：

bash
docker-compose -f docker-compose.yml -f docker-compose.apps.yml up --build

启动组件包括源/目标Event Store、Localstack、Seq日志工具、Ditto默认版及Kinesis版。

测试复制

通过API在源集群创建事件：

bash
curl --location --request POST 'http://localhost:2113/streams/customer-6d3d4dca-9889-416b-a266-4bf760261f01' \
-u admin:changeit \
-H 'Content-Type: application/vnd.eventstore.events+json' \
--data-raw '[
    {
        "eventId": "ab932527-5b9a-4868-82f3-42194bbfd022",
        "eventType": "customer_registered",
        "data": {
            "first_name": "John",
            "last_name": "Smith",
            "phone_number": "0111111111111"
        },
        "metadata": {
            "source": "ditto"
        }
    }
]'

创建后可在源集群查看事件，并在目标集群/Kinesis验证复制结果。