Kafka跨集群数据同步方案对比：MirrorMaker 2.0、Connect与自研管道实战

从"把Kafka数据同步到另一个Kafka"这个需求进入正文。我自己最早接触这件事的时候，第一反应也是"消费一条、生产一条"那么简单，直到某次数据对不上账，从凌晨排查到天亮，才意识到这里面的坑比想象中多得多。同步不是搬运工，不是简单的读写就能交差的。

这篇文章我把自己在几个项目里实际用过的Kafka到Kafka同步方案整理出来，包括官方的MirrorMaker 2.0、基于Kafka Connect的自定义转换管道，以及纯自研的消费生产管道。场景覆盖跨机房容灾、集群迁移、多活拓扑和边缘集群向中心汇聚这几类最常见的需求，适合正在做Kafka集群规划、或者正在为跨集群复制选型的同学参考。

1. 开始动手前先想清楚：你同步的到底是一堆消息，还是一整套语义

很多人第一次写同步脚本时，脑子里只有"消息体"，以为把value搬过去就完事了。实际上Kafka的topic是一个有状态的对象，分区数量、副本因子、消息key的哈希路由规则、topic级别的retention配置、压缩类型，甚至ACL权限，全都是"数据"的一部分。

举个例子，源集群某个topic有12个分区，目标集群建topic时默认只建了3个分区，消息同步过去的key哈希落点全部变了。对生产者和消费者来说，这不仅仅是分区数不同的问题，而是同一个key可能落在不同分区上，下游依赖分区顺序做状态恢复的程序直接废掉。所以我每次做同步方案之前，都会先做一次"topic元数据对账"，把分区数、副本数、cleanup.policy、compression.type这些字段全部拉出来逐项比对，然后再谈消息怎么搬。

另一个容易被忽略的是offset语义。同步工具往目标集群写数据时，目标集群里的offset是从0开始自然递增的，源集群里的offset可能是几千万。如果下游任务依赖于"aligned offset"，比如做跨集群的lag对比，就需要在方案里明确offset映射关系。MirrorMaker 2.0会在目标topic的消息头上写入源集群的offset和分区信息，但那些只是辅助标记，目标topic自己的offset依然是从0开始的。对这个机制不理解，排障的时候会走很多弯路。

再聊聊拓扑。同步方向直接决定方案选型：

• 单向灾备：源集群为主，目标集群只做备份，数据不允许回流，这是最简单的场景，MirrorMaker 2.0默认配置就能满足。
• 双向互备：两个集群同时读写，数据互相复制，最麻烦的问题就是消息回环，A集群的消息同步到B集群，B集群又同步回A集群，形成死循环。官方默认的复制策略会给topic加集群别名前缀来规避回环，但代价是目标端topic名变了。
• 汇聚：多个边缘集群把数据往中心集群同步，相当于多对一。
• 分发：中心集群把数据推送到各个边缘集群，是一对多。

这四种拓扑对同一款工具的配置要求完全不同。所以我一般建议在选工具之前，先用一张拓扑图把数据流向画清楚，标出哪些topic是单向的、哪些是双向的，再考虑配置和改造。直接上手配MirrorMaker，往往会在双向同步这块栽跟头。

2. 方案一：MirrorMaker 2.0 — 官方自带，配置一次能用很久

MirrorMaker 2.0（后面简称MM2）在很多人的印象里还是老版MirrorMaker的升级版，其实它内部完全重写了，底层直接跑在Kafka Connect框架上。核心组件是内置的一组连接器，一个source connector负责从源集群消费，一个sink connector负责把数据写入目标集群，MM2本身负责管理这两个连接器的生命周期、offset检查点和心跳。

2.1 先弄明白它怎么工作的

MM2在复制数据的同时，会在目标集群里创建一组内部topic，用来记录复制进度和链路健康状态。

• heartbeat：模拟一条持续发送的心跳消息，用来检测源集群和目标集群之间的同步链路是否健康。
• checkpoint：定期把消费进度（offset）记录下来，这样即使MM2重启，也能从最近的checkpoint继续复制，不用从头消费。
• offset-sync：记录源集群offset和目标集群offset的映射关系。

这些内部topic的名字里通常带有源集群和目标集群的别名。如果你的集群没必要做双向复制，这些内部topic直接使用默认配置就行，但如果你需要双向复制，这些内部topic本身也需要避免在两个集群间被循环复制回去，这时就要靠复制策略来控制。

2.2 一份能跑起来的配置长什么样

我贴一份自己常用在单向灾备场景下的MM2配置，做了脱敏处理：

# 定义集群别名 clusters = source, target # 源集群连接信息 source.bootstrap.servers = 10.0.0.11:9092,10.0.0.12:9092,10.0.0.13:9092 source.security.protocol = SASL_PLAINTEXT source.sasl.mechanism = SCRAM-SHA-256 # 目标集群连接信息 target.bootstrap.servers = 10.0.1.11:9092,10.0.1.12:9092,10.0.1.13:9092 target.security.protocol = SASL_PLAINTEXT target.sasl.mechanism = SCRAM-SHA-256 # 开启从源集群到目标集群的同步 source->target.enabled = true source->target.topics = order.*, user.*, inventory.* # 自动发现新topic的时间间隔，默认是60秒 source->target.refresh.topics.interval.seconds = 30 # checkpoint写入频率，建议不要低于10秒 source->target.emit.checkpoints.interval.seconds = 15 target.checkpoints.topic.replication.factor = 3 target.heartbeats.topic.replication.factor = 3 # 同步topic的ACL？默认false，建议保持false，ACL单独维护 source->target.sync.topic.acls.enabled = false # 复制策略，默认会在目标端创建 "source.order_xxx" 格式的topic replication.policy.class = org.apache.kafka.connect.mirror.DefaultReplicationPolicy

这些参数里最容易被忽略的是三个replication.factor相关的配置。很多测试环境broker就1台，副本因子默认1无所谓；到了生产环境，如果只复制数据没复制这些内部topic的副本因子，心跳topic和checkpoint topic会变成单副本，哪台broker挂了整个同步任务就停摆。我见过不止一次这种事故，配置里写清楚target.checkpoints.topic.replication.factor = 3，基本可以避免这类问题。

还有source->target.enabled这个参数，它其实是source和target两个方向配置的前缀。如果集群有4个，要同步的链路有6条，每一条都写一段这样的前缀配置，配置文件会非常长。我建议把公共的认证信息抽到前面的集群配置里，只把方向相关的source->target.*单独维护。

2.3 用起来最需要注意的几个点

第一，版本匹配。MM2的Kafka版本最好与源集群broker版本保持大版本兼容，至少不能差太多。我曾经在源集群还是老版本broker的时候，用了新版本自带的MM2，结果在读取一些老版本独有的元数据时出现兼容性问题，消费线程频繁报错，后来把MM2的客户端版本降到与源集群接近才恢复正常。

第二，topic数量很多的时候，正则表达式要慎重。source->target.topics支持逗号分隔多个正则，比如order.*, user.*，但实际使用时，正则匹配是完整匹配，所以order.*不会匹配my_order_detail，这点和很多人想的"包含匹配"不一样。如果想匹配所有topic，直接写.*即可，但要确认内部topic不会被自己复制成无线循环。

第三，千万注意目标集群的topic容量。MM2默认会在目标集群自动创建topic，但创建的topic分区数和源集群是一致的，这个设计很贴心。不过如果目标集群broker数量比源集群少，自动创建的副本因子默认是3，副本数可能超过broker数量，topic会创建失败。

第四，双向复制时不要轻易改复制策略。官方默认的DefaultReplicationPolicy会在目标端给topic名字加源集群别名前缀，比如源集群叫A，topic叫order_paid，同步到B集群后名字变成A_order_paid。这样虽然规避了回环，但下游消费者必须感知新topic名。有人为了保持名字不变，自定义了复制策略直接透传topic名，这在双向同步场景下很容易造成无限循环，必须配合消息头过滤和链路标记一起改造，复杂度暴涨。对大多数团队来说，用官方默认策略反而最省心。

3. 方案二：Kafka Connect + 自定义Sink连接器，复制数据的同时顺手做加工

MM2解决的问题是"原样复制"，但很多业务场景不满足于原样复制。我遇到过几种典型需求：同步过程中把Avro格式的消息转成JSON方便目标端使用；过滤掉某些字段或者某些事件类型；根据消息内容把单topic按业务键拆分到目标端多个topic；甚至需要在同步时给消息打上数据中心标签，方便下游识别数据来源。

这些需求用MM2做就很别扭，而Kafka Connect自定义Sink连接器是更适合的落地方式。

3.1 思路和架构

Kafka Connect本身的source connector还是从源集群消费，但sink连接器不是"直接写目标集群"，而是交给我们自己实现的SinkTask代码，在Task里完成转换、过滤、拆分，再通过一个内部创建的Kafka Producer写入目标集群。这样做的最大好处是，Kafka Connect框架替我们管理了offset的持久化和断点续传，我们只需要专注处理"一条record怎么转换、怎么发送"这个逻辑。

整个管道的结构是：

源集群 -> Kafka Source Connector -> Connect框架(自动提交offset) -> 自定义SinkTask -> 转换/过滤/拆分 -> 内部Producer -> 目标集群

这种方案比MM2灵活，但也意味着我们失去了MM2内置的链路心跳和checkpoint管理。好在Kafka Connect框架本身就提供了rest API，我们可以通过API查询connector的状态和lag，配合监控工具做告警。

3.2 一个最小可用的SinkTask代码骨架

我贴一段自己项目里脱敏后的简化版实现，用的Java语言：

public class TransformSinkTask extends SinkTask { private KafkaProducer<String, byte[]> producer; private String targetTopicPrefix; @Override public void start(Map<String, String> props) { Properties producerProps = new Properties(); producerProps.put("bootstrap.servers", props.get("target.bootstrap.servers")); producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); producerProps.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer"); // 幂等生产者，防止网络重试导致目标端重复消息 producerProps.put("enable.idempotence", "true"); producer = new KafkaProducer<>(producerProps); targetTopicPrefix = props.getOrDefault("target.topic.prefix", ""); } @Override public void put(Collection<SinkRecord> records) { for (SinkRecord record : records) { String key = (String) record.key(); byte[] value = (byte[]) record.value(); // 这里可以加过滤逻辑，比如跳过某些无意义的系统消息 if (value == null || value.length == 0) { continue; } // 这里可以加转换逻辑，比如解包嵌套字段打上数据中心标签 // 也可以加拆分逻辑，根据key的前缀路由到不同目标topic String targetTopic = targetTopicPrefix + record.topic(); producer.send(new ProducerRecord<>(targetTopic, key, value)); } // 本批数据全部发送完成后，Kafka Connect框架才会提交offset // 如果发送是异步的，必须在put方法返回前flush，否则会丢失消息 producer.flush(); } @Override public void stop() { if (producer != null) { producer.close(); } } }

这段代码看着简单，但几个细节直接决定数据会不会丢。

最关键的细节是在put方法末尾必须调producer.flush()。send()是异步的，如果只调用send()就返回，Kafka Connect会认为这条记录处理成功了并提交offset，但消息可能还在内部缓冲区，进程一崩就没了。我第一次写的时候就踩了这个坑，同步任务运行两周后目标端发现约0.1%的消息丢失，排查了很久才发现是没加flush。

另一个细节是SinkRecord中的value可能是多种类型，不只是byte[]。如果源topic用的是String序列化器，value就是String，直接用(byte[])强转会抛异常。我在生产环境里一般同时判断record.valueType()再处理，或者统一在source connector配置里设置value.converter为ByteArrayConverter。

3.3 错误处理与断点续传

Kafka Connect框架的offset提交跟SinkTask的put方法是同步关系：只有put正常返回，框架才认为这一批数据成功。如果put抛出异常，框架会停掉这个task，不断重试。这意味着我们自己的同步代码里，网络抖动、目标集群不可用等情况，到底要不要重试，重试多久，一定要考虑清楚。

最简单的策略是所有发送失败都抛异常，让框架停掉task，等网络恢复后自动重启。这种做法稳定性最好，不会丢数据，但会遇到一个麻烦：如果某条消息是脏数据、目标集群的所有重试都失败，task会一直卡死在这里，后续所有消息全部阻塞。

稍微复杂点的策略是把失败消息写入一个内部的"死信topic"，从当前批次里移除，继续处理后面的记录。这样丢掉的只是死信本身，且我们能从死信topic里人工恢复。Kafka Connect社区里有一些现成的死信处理器，但集成到自定义SinkTask里也不复杂，就是在catch块里把失败记录发送到一个单独的dlq-${topic}，然后记录一下现场信息。这个策略在数据一致性要求不那么极端的项目里非常实用。

我自己现在做的项目是两种策略结合：可重试的异常（比如网络超时、broker不可用）直接抛出，让框架重试；不可重试的异常（比如反序列化失败、字段缺失）走死信topic。运维上会盯死信topic的消息量，一旦超过阈值就触发告警。

4. 方案三：自研消费者-生产者管道，灵活度最高但工作量也最大

如果场景很简单，比如一次性的集群迁移、临时同步几个topic，根本不想部署一套Kafka Connect集群，自研一个消费-生产管道是最快的方案。这种管道本质上就是一个普通的Java或者Python进程，一边从源集群的consumer消费，一边用producer往目标集群发送。

优点很直接：没有额外组件，代码量不大，出了问题能看清每一行逻辑。缺点也很明显：需要自己处理消费进度、幂等、事务、重平衡、端到端一致性这些在成熟框架里已经被实现好的东西。小规模场景用得很舒服，规模一大就处处是坑。

4.1 最简单的版本：手动提交offset

我见过很多同步脚本是这么写的：从源集群订阅topic，设置enable.auto.commit=true，然后收到一条发一条。这种写法在生产环境基本等于埋雷。

如果不要事务，我建议至少改成手动提交offset，核心逻辑是：

1. 关闭自动提交：enable.auto.commit=false。
2. 调用poll()拿到一批消息。
3. 把这一批消息全部发送到目标集群，并且producer.flush()，确认这一批真的发送完成。
4. 再调用commitSync()提交这批消息的offset。
5. 如果中途任何一步失败，不提交offset，下次启动时就会从上次提交的位置重新消费。

这样做的原理很朴素：先确保数据真的写到目标端了，再移动源端的消费进度。它保证的是"至少一次"语义——崩溃时消息可能被重复消费。配合目标端producer开启幂等（enable.idempotence=true），重复消费时目标端也不会产生重复消息，这是最实用的组合。

4.2 进阶：用事务实现精确一次

如果业务上要求源端和目标端强一致，不允许任何重复，那就需要把这个管道升级为Kafka的consume-transform-produce事务模型。

基本思路是：producer开启事务，消费一批消息之后，在同一事务里执行"向目标topic发送转换后的消息"和"向源集群提交offset"两个操作。这两个操作要么同时成功，要么同时失败，不会出现"消息写过去了但offset没提交"这种状态。

核心代码逻辑如下：

// 初始化事务 producer.initTransactions(); while (true) { ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(1000)); if (records.isEmpty()) { continue; } // 开启事务 producer.beginTransaction(); try { for (ConsumerRecord<String, byte[]> record : records) { // 这里处理转换逻辑 byte[] transformed = transform(record.value()); producer.send(new ProducerRecord<>(targetTopic, record.key(), transformed)); } // 把源集群的消费位移提交到事务中 producer.sendOffsetsToTransaction( buildOffsetMap(records), new ConsumerGroupMetadata("my-sync-group") ); // 提交事务 producer.commitTransaction(); } catch (Exception e) { // 出现异常则回滚整个事务 producer.abortTransaction(); throw e; } }

用事务机制之后，消费者侧需要设置isolation.level=read_committed，否则会读到未提交事务里的消息。sendOffsetsToTransaction这个方法要求传入ConsumerGroupMetadata，也就是消费组的元数据，所以需要给同步管道单独设置一个稳定的group.id，别和别的应用混用。

事务机制的代价是性能会下降，因为每一次commit都要经过一轮生产者事务协调。但这个模型值得一写，因为在"跨集群复制"这种场景下，精确一次的语义很值钱，不值得为了一点吞吐牺牲一致性。

4.3 普通人最容易忽略的坑

第一个坑是重平衡。消费者组如果还有其他消费者实例在跑，当实例宕机或扩容时，分区会被重新分配，这时正在处理的消息可能还没提交offset。事务模型可以缓解，但非事务模型下，重平衡期间正在消费的消息可能丢失。简单场景下我建议消费组就一个实例，不做负载均衡，减少重平衡带来的不确定性。

第二个坑是消息顺序。源topic如果是12个分区，目标topic也应该是12个分区，并且发送的时候不能做任何key变化，否则同一个key的消息会落在不同分区，顺序就乱了。多线程发送比单线程快，但你无法保证多个线程之间的顺序，所以如果你对顺序敏感，建议源端到目标端启用单线程生产者模型，用吞吐换顺序。

第三个坑是心跳线程。消费者在poll()间隔太长时会被踢出消费组。如果你在poll()循环里做了耗时的转换逻辑或者调用了外部接口，耗时会直接影响心跳间隔。解决办法是调大max.poll.interval.ms，或者把poll循环和业务处理拆到不同线程，但后者会引入并发顺序问题，代价不小。我自己做同步管道时，处理逻辑都很轻量，坚决不在poll线程里做重IO操作。

5. 方案选型：到底该用哪个，以及几条能少踩坑的实战经验

四种思路讲完了，下面把MM2、Kafka Connect自定义Sink、自研管道这几个方案放在一起，从学习和运维成本、同步保序能力、灵活性这几个维度直接比较。

选哪个不是越高级越好，而是看你的核心约束是什么。

如果重点是稳定和快速落地，目标只是"把topic原样复制过去"，MM2是默认选择。你不需要写代码，配置也简单，链路健康检查都已经做好，省心程度最高。

如果同步的同时想做转换、过滤，或者源和目标的消息格式不兼容，需要做字段映射，那就用Kafka Connect自定义Sink。它保留了管理offset的框架能力，又给了你加工数据的自由度。

如果只是临时同步一两个topic，不想引入任何框架和运维组件，自研管道反而是最务实的方案。但一定记住把幂等生产者打开，把自动提交关掉，把flush写在正确的位置。

5.1 端到端延迟和吞吐怎么调

我自己的经验是，跨集群同步的瓶颈很少在Kafka本身，更多在转换逻辑和网络链路上。

一条消息从源集群写入到目标集群可消费，端到端延迟基本由四部分组成：源端poll间隔、转换耗时、producer批量发送等待时间、目标端确认时间。很多人同步时发现延迟高，但本地测试发送很快，问题几乎都出在poll间隔上。如果你把poll(Duration.ofMillis(1000))写死成1秒，那即使目标集群秒回，端到端延迟也至少多了1秒。我一般把poll间隔调到100到200毫秒，兼顾CPU和延迟。

吞吐上，调整batch.size和linger.ms是立竿见影的手段。如果linger.ms=0，producer会立即发送，吞吐上不去；如果调到5到10毫秒，每条消息会攒一小批再发，吞吐能提升不少，但对延迟影响不大，因为5毫秒在跨集群同步场景里几乎可以忽略。

5.2 版本兼容和消息格式上的几个坑

我做同步方案时遇到过几个高频问题，直接列出排查思路：

还有一个容易忽略的点是topic的cleanup.policy。源topic是compact策略，同步到目标端如果默认变成delete策略，语义就变了，下游按key查聚合数据的逻辑可能出错。我在同步配置里都会显式设置目标topic的cleanup.policy和源端一致。

5.3 监控和验收的经验

同步链路跑起来之后，日常监控不能只看两边的消息条数。最有效的手段是双端同时落一条探针消息，比如每分钟在源集群发送一条带独立ID的测试消息，在目标集群消费端统计到达时间，计算出端到端延迟。延迟超过阈值就告警。

做一致性验收时，不建议只做条数对比。因为业务可能对某些消息做过过滤，数量本来就不完全相等。更好的做法是给消息体加一个全链路唯一的业务ID，在源和目标分别生成同一批ID的摘要，比对摘要是否一致。摘要一致，基本可以说明内容没有变形。

我自己做验收时会在源集群创建一个专门的对账topic，写入一批预先生成的测试数据，记录目标端按顺序收到的内容，然后比对hash。这个方法跑了几年，虽然笨，但比看监控面板更踏实。

最后分享一点自己的体会

这几个方案我都在真实场景里跑过，也踩过不少坑。如果只能记住一句话，我想说：偏移量的一致性比数据本身的一致性更重要。数据搬运只是表面，offset管理才是整个同步链路的地基。无论是用MM2的checkpoint，还是自己维护消费位移，都要想清楚"源端消费进度"和"目标端写入结果"之间到底是怎么对齐的。这个想明白了，其他问题都能顺着排查；这个没想明白，调再多的参数也是在晃地基。

最后再补充一个实用技巧：不管用哪种方案，在目标集群上给同步管道单独建一个消费组，别和业务应用混用同一个group.id。这样出问题时，你能用Kafka自带的工具看清楚同步任务的消费进度和业务消费进度分别在哪里，能省下不少排查时间。跨集群同步这件事，做得好没有人注意到你，做得不好半夜电话一定到。谨慎一点，总没错。