kafka中MirrorMaker1和MirrorMaker2的区别

kafka中MirrorMaker1和MirrorMaker2的区别

引用

CSDN

1.

https://blog.csdn.net/m0_37899908/article/details/143259158

kafka中MirrorMaker1和MirrorMaker2的区别

1. 工作原理与架构

MM1：

• 本质上是Kafka的消费者（Consumer）和生产者（Producer）的结合体。
• 通过消费源集群的数据，并将其生产到目标集群，实现数据的复制和同步。
• 使用了Kafka client的high level API。

MM2：

• 基于Kafka Connect框架开发，利用了Kafka Connect的source connector和sink connector。
• Source connector负责读取远端数据中心的数据，sink connector负责将拉回来的数据写入本地的数据中心。
• 提供了更高级的同步机制，如自动探测新topic、自动同步topic配置和ACL（访问控制列表）等。

2. 功能与特性

MM1：

• 功能相对简单，主要实现数据的复制和同步。
• 缺乏监控手段，无法实时了解同步任务的状态和性能。
• 在处理active-active场景时，需要手动配置来解决循环同步问题。

MM2：

• 提供了丰富的监控指标和管理接口，可以实时监控复制任务的状态和性能。
• 支持跨版本兼容，可以方便地在不同版本的Kafka集群之间进行数据复制和同步。
• 提供了更高的灵活性和可扩展性，可以根据实际需求进行配置，满足不同的数据同步需求。
• 支持active-active集群对以及任意数量的active集群，支持跨IDC同步、aggregation和其他复杂拓扑。

3. 部署与维护

MM1：

• 部署相对复杂，需要为每个需要同步的Kafka集群对配置一个MirrorMaker实例。
• 在维护方面，需要定期检查和调整配置，以确保同步任务的正常运行。

MM2：

• 部署更加简单，可以通过Kafka Connect集群来管理多个同步任务。
• 提供了更简洁的配置方式，降低了维护成本。

4. 性能与稳定性

MM1：

• 在网络不稳定或负载较高的情况下，可能会遇到频繁的rebalance问题，影响同步的稳定性和效率。

MM2：

• 采用了更先进的同步机制，避免了频繁的rebalance问题。
• 提供了更高的吞吐量和更低的延迟，适用于大数据场景下的数据同步。

MM1局限性

• 静态的黑名单和白名单
• topic信息不能同步
• 必须通过手动配置来解决active-active场景下的循环同步问题
• rebalance导致的性能问题
• 缺乏监控手段
• 无法保证Exactly Once
• 无法提供容灾恢复

kafka带来的MM2

1. MM2简介

kafka开源社区也终于在kafka2.4带来了自己的企业级解决方案MirrorMaker-V2(MM2)。MM2修复了MM1所存在的局限性。

MM2是基于kafka connect框架开发的。与其它的kafka connecet一样MM2有source connector和sink connetor组成，不熟悉kafka connect概念的同学可以这样认为，source connector就是MM1中的消费者，它负责读取远端数据中心的数据。sink connetor是生产者，他负责将拉回来的数据写入本地的数据中心。

MM2共含有4种类型的connector：

• MirrorSourceConnector
• MirrorSinkConnector
• MirrorCheckpointConnector
• MirrorHeartbeatConnector

与MM1不同的是，MM2的source和sink两个connector包含了，源数据的消费者，远端数据的生产者，和一对AdminClient用来同步topic配置信息。

它的部署方式跟MM1相同，都是部署在目标集群方。

2. MM2具有如下特性:

• 基于Kafka Connect framework和生态
• 自动探测新topic, partition
• 自动同步topic配置，自动同步topic acl
• 支持active-active集群对，以及任意数量的active集群
• 支持跨IDC同步， aggregation和其他复杂拓扑
• consumer offset等meta信息的同步和翻译
• no rebalance，减少同步波动
• 提供广泛的指标，例如跨多个数据中心/集群的端到端复制延迟
• 容错和水平可扩展

3. MM2配置详解：

#定义集群别名
clusters = A, B
A.bootstrap.servers = A_host1:9092, A_host2:9092, A_host3:9092 # 设置A集群的kafka地址列表
B.bootstrap.servers = B_host1:9092, B_host2:9092, B_host3:9092 # 设置B集群的kafka地址列表
A->B.enabled = true # 开启A集群向B集群同步
A->B.topics = .* # 允许同步topic的正则
B->A.enabled = true # 开启B集群向A集群同步
B->A.topics = .* # 允许同步topic的正则
#MM2内部同步机制使用的topic，replication数量设置
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1
#自定义参数
sync.topic.configs.enabled=true #是否同步源topic配置信息
sync.topic.acls.enabled=true #是否同步源ACL信息
emit.heartbeats.enabled=true #连接器是否发送心跳
emit.heartbeats.interval.seconds=5 #心跳间隔
emit.checkpoints.enabled=true #是否发送检查点
refresh.topics.enabled=true #是否刷新topic列表
refresh.topics.interval.seconds=5 #刷新间隔
refresh.groups.enabled=true #是否刷新消费者组id
refresh.groups.interval.seconds=5 #刷新间隔
readahead.queue.capacity=500 #连接器消费者预读队列大小
replication.policy.class=org.apache.kafka.connect.mirror.DefaultReplicationPolicy #使用LegacyReplicationPolicy模仿MM1
heartbeats.topic.retention.ms=1 day #首次创建心跳主题时，设置心跳数据保留时长
checkpoints.topic.retention.ms=1 day #首次创建检查点主题时，设置检查点数据保留时长
offset.syncs.topic.retention.ms=max long #首次创建偏移量主题时，设置偏移量数据保留时长
replication.factor=2 #远端创建新topic的replication数量设置

4. MM2启动

bin/connect-mirror-maker.sh config/connect-mirror-maker.properties