4. Kafka跨集群复制：MirrorMaker 2.0架构、配置、监控与调优

跨集群复制，说白了就是让两个Kafka集群之间数据同步。你可能会问：为什么需要这个？我举个例子，你在北京机房部署了一套Kafka集群，上海机房也有一套。如果北京机房挂了，上海那边得能顶上。或者你想做异地多活，让用户就近读写。这时候，MirrorMaker 2.0就派上用场了。

我个人习惯把MM2看作一个「数据搬运工」。它不像1.0版本那么简陋，2.0直接基于Kafka Connect框架，自带了很多高级功能。嗯，咱们今天就把它的架构、配置、监控和调优掰开揉碎了讲清楚。

核心要点： MirrorMaker 2.0 不是简单的数据拷贝工具，它解决了跨集群复制中的拓扑管理、偏移量同步、连接器自动创建等关键问题。

4.1 MirrorMaker 2.0 架构解析

先看一张整体架构图，你就能明白MM2是怎么工作的了。

MirrorMaker 2.0 跨集群复制架构 源集群 (Source) Topic-A (分区0,1,2) Topic-B (分区0,1) 偏移量主题 心跳主题 目标集群 (Target) Topic-A (分区0,1,2) Topic-B (分区0,1) 偏移量主题 心跳主题 MirrorMaker 2.0 (Kafka Connect) MirrorSourceConnector MirrorCheckpointConnector MirrorHeartbeatConnector 说明： MM2 运行在独立的 Kafka Connect 集群上，通过三个内置连接器完成数据复制、偏移量同步和心跳检测。

这张图展示了MM2的核心工作方式。源集群和目标集群各自独立运行，MM2作为中间层，通过三个内置连接器完成复制任务。我刚开始接触MM2时，总觉得它就是个高级版的消费者-生产者，后来才发现它远不止这么简单。

4.2 三个核心连接器

MM2内部有三个连接器，各司其职。咱们一个一个看。

连接器名称 作用 关键配置
MirrorSourceConnector 从源集群消费数据，写入目标集群 topics, replication.factor, refresh.topics.interval.seconds
MirrorCheckpointConnector 同步消费者偏移量，实现灾备切换 checkpoints.topic.replication.factor, emit.checkpoints.interval.seconds
MirrorHeartbeatConnector 发送心跳，检测集群连通性 heartbeats.topic.replication.factor, emit.heartbeats.interval.seconds

你想想看，如果没有偏移量同步，灾备切换时消费者就得从头消费，那数据量大的时候根本扛不住。MirrorCheckpointConnector就是解决这个问题的。

我的经验： 我曾经在一个项目中只配了MirrorSourceConnector，结果切换时消费者偏移量全乱了，数据重复消费了好几百万条。从那以后，我每次都会把三个连接器都配齐。

4.3 配置实战

配置MM2其实不复杂，但有几个坑得注意。先看一个完整的配置文件示例。

# mm2.properties
clusters = source, target

# 源集群配置
source.bootstrap.servers = source-kafka1:9092,source-kafka2:9092
source.security.protocol = SASL_SSL
source.sasl.mechanism = PLAIN
source.sasl.jaas.config = org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

# 目标集群配置
target.bootstrap.servers = target-kafka1:9092,target-kafka2:9092
target.security.protocol = SASL_SSL
target.sasl.mechanism = PLAIN
target.sasl.jaas.config = org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

# 复制方向：从source到target
source->target.enabled = true

# 复制哪些主题
source->target.topics = .*
# 排除内部主题
source->target.topics.exclude = .*\.internal,.*\.__consumer_offsets

# 连接器配置
source->target.replication.factor = 3
source->target.refresh.topics.interval.seconds = 60
source->target.sync.group.offsets.enabled = true
source->target.sync.group.offsets.interval.seconds = 30

# 心跳配置
heartbeats.topic.replication.factor = 3
emit.heartbeats.interval.seconds = 5

# 检查点配置
checkpoints.topic.replication.factor = 3
emit.checkpoints.interval.seconds = 10

配置里有个关键点：source->target.topics.exclude。我建议一定要把内部主题排除掉，否则会把Kafka的内部元数据也复制过去，容易出问题。

4.4 监控指标

MM2运行起来后，怎么知道它干得好不好？监控是关键。我个人习惯关注以下几个指标。

指标名称 MBean路径 说明 告警阈值
复制延迟 kafka.connect.mirror:type=MirrorSourceConnector,source=*,target=* 源集群最新消息与目标集群已复制消息的时间差 > 10秒
字节率 kafka.connect.mirror:type=MirrorSourceConnector,source=*,target=* 每秒复制的字节数 根据业务基线
心跳延迟 kafka.connect.mirror:type=MirrorHeartbeatConnector,source=*,target=* 心跳消息从发送到接收的延迟 > 5秒
连接器状态 kafka.connect:type=connector-metrics,connector=* 连接器是否正常运行 非RUNNING

你可能会问：这些指标怎么采集？我一般用Prometheus + JMX Exporter。把JMX Exporter挂到MM2进程上，Prometheus定期拉取，Grafana展示。嗯，这套组合拳打下来，MM2的运行状态一目了然。

注意： 复制延迟这个指标要特别关注。如果延迟持续增大，说明MM2的处理能力跟不上数据生产速度。这时候就需要调优了。

4.5 调优策略

调优这事儿，说白了就是找到瓶颈然后解决它。我总结了几条实战经验。

4.5.1 并行度调优

MM2的并行度由两个参数控制：tasks.max 和 replication.policy.separator。tasks.max决定了每个连接器启动多少个task，task越多并行度越高。

# 增加并行度
source->target.tasks.max = 10
# 每个task处理的分区数
source->target.partition.assignment.strategy = org.apache.kafka.connect.mirror.DefaultReplicationPolicy

我曾经遇到过一个场景：源集群有100个分区，tasks.max只设了2，结果复制延迟飙到30秒。后来把tasks.max调到10，延迟降到3秒以内。你想想看，这就是并行度的威力。

4.5.2 内存调优

MM2基于Kafka Connect，内存主要消耗在消费者和生产者缓冲区。我建议这样配：

# 消费者缓冲区
source.consumer.fetch.max.bytes = 52428800  # 50MB
source.consumer.max.partition.fetch.bytes = 10485760  # 10MB

# 生产者缓冲区
target.producer.buffer.memory = 67108864  # 64MB
target.producer.batch.size = 262144  # 256KB
target.producer.linger.ms = 100

# JVM堆内存
KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

这里有个小技巧：linger.ms不要设太大，否则会增加延迟。100ms是个不错的起点，你可以根据业务对延迟的容忍度调整。

4.5.3 网络调优

跨集群复制，网络往往是最大瓶颈。我建议：

  • 使用专线连接两个集群，避免公网抖动
  • 开启压缩：target.producer.compression.type = snappy
  • 调整TCP参数：socket.send.buffer.bytes 和 socket.receive.buffer.bytes 设为1MB

核心原则： 调优不是一蹴而就的。先监控，再分析，最后调整。每次只改一个参数，观察效果后再改下一个。

4.6 避坑指南

做跨集群复制，坑不少。我踩过的几个，分享给你。

  • 主题命名冲突： 默认情况下，MM2会在目标集群创建 source-cluster-name.topic-name 格式的主题。如果你不想加前缀，可以设置 replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy。
  • 偏移量不同步： 我曾经遇到过消费者组偏移量没同步的情况，原因是 sync.group.offsets.enabled 没开。记得把它设为true。
  • 连接器重启： MM2连接器挂了怎么办？我建议用systemd或Kubernetes管理MM2进程，设置自动重启策略。
  • 版本兼容性： 源集群和目标集群的Kafka版本最好一致。如果不一致，至少保证大版本相同，比如都是2.x或3.x。

嗯，这些坑你提前知道，就能少走很多弯路。我记得有一次生产环境出了故障，就是因为版本不一致导致序列化问题，排查了好几个小时才找到原因。

4.7 总结

MirrorMaker 2.0是Kafka跨集群复制的利器。它的架构清晰，三个连接器各司其职。配置时注意排除内部主题，监控时重点关注复制延迟，调优时从并行度、内存、网络三个维度入手。避坑指南里的几个点，你最好记下来。

说白了，MM2用好了，你的Kafka灾备体系就稳了一半。剩下的，就是结合Flink做端到端的灾备集成了。

← 上一章 📖 返回目录 下一章 →