因为Go生态缺乏统一的数据同步框架，现成库仅支持单点通信，无法满足跨源联动、断点续传、并发控制与失败重试等业务需求；自研框架可收口checkpoint、worker pool、backoff retry等核心逻辑，通过Source/Transformer/Sink三层解耦实现可靠、可控、可扩展的至少一次同步。
为什么不用现成的同步库，而要自己写框架
Go 生态里没有统一的“数据同步框架”，gocql、pglogrepl、redis-go 各管一摊，但业务常需要跨源联动：比如 MySQL binlog 变更 → 写入 Elasticsearch + 推送 Kafka。官方库只负责单点通信，不处理状态跟踪、断点续传、并发冲突、失败重试这些事。自己搭框架不是为了造轮子，是把 checkpoint、worker pool、backoff retry 这些逻辑收口，避免每个项目重复写 goroutine 控制和错误兜底。
核心组件怎么组织：Source / Transformer / Sink
这个三层结构不是为了分层而分层，而是为了解耦变更捕获、业务加工、目标投递三类关注点：
-
Source负责拉取原始变更（如监听 MySQL 的binlog.Streamer，或轮询 PostgreSQL 的pglogrepl.StartReplication），必须支持ResumeFrom(position)接口，否则断连后无法续同步 -
Transformer是可选中间层，做字段映射、过滤、拼接（比如把user_id和order_id合成复合键），它不能阻塞，也不能带状态——否则并发消费时会错乱 -
Sink必须实现WriteBatch([]*Event) error和Ack(position)，且Ack必须在WriteBatch成功后才调用，否则丢数据
示例中 Sink 对 Kafka 的封装要特别注意：sarama.SyncProducer 默认不保证顺序，得配 Partitioner: sarama.NewHashPartitioner 并确保 key 不为空；对 ES 则要禁用 refresh=true，改用 ?refresh=wait_for 避免批量写被拆成多次刷新。
如何保证至少一次（at-least-once）不丢数据
关键不在“怎么发”，而在“发完怎么确认”：
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器，提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象，支持命名依赖等
立即学习“go语言免费学习笔记（深入）”；
- 所有
Source必须提供幂等 position（MySQL 用filename + position，PostgreSQL 用LSN，Kafka 用topic/partition/offset） -
Sink.WriteBatch成功后，立刻调用checkpoint.Store(position)，存储介质必须和 Sink 同可用区（比如 Sink 写 AWS us-east-1 的 Kafka，checkpoint 就不能存 us-west-2 的 S3） - 失败时不要直接跳过事件，而是把
Event+ 当前position写入本地retry.db（用bolt或badger），由单独 goroutine 按指数退避重试 - 禁止在
Transformer里调time.Sleep或阻塞 IO，这类操作会让整个 worker 卡住，导致 checkpoint 停滞
一个典型坑：checkpoint.Store 如果用文件写入，没加 os.O_SYNC 标志，机器掉电可能丢失最后几条 position，重启后重复消费。
并发模型与资源控制怎么做才不爆内存
默认用 runtime.GOMAXPROCS(0) 不够，得按数据源吞吐动态调：
- 每个
Source启一个 goroutine 拉取，用chan *Event向下游推送，buffer size 设为 1024～4096，太小易堵，太大吃内存 -
Transformer和Sink共享同一个worker pool（推荐panjf2000/ants），池大小 = CPU 核数 × 2，避免创建过多 goroutine - 对高吞吐场景（如每秒万级 event），
Sink.WriteBatch的 batch size 要和网络 MTU 对齐：Kafka 设 1MB，ES 设 50 条/批，Redis pipeline 设 100 条，硬凑大 batch 反而因单次超时失败率上升 - 所有
Event结构体禁止含指针字段（如*string），避免 GC 扫描开销；时间戳统一用int64存 UnixMilli，别用time.Time
真正难的不是写通路，是当 MySQL 主从延迟飙升时，如何让 Sink 不疯狂积压内存——得在 channel 上游加 semaphore.Acquire()，拿不到信号量就 drop 或降级到磁盘暂存。
