什么是 MirrorMaker？

MirrorMaker是Kafka附带的一个用于在Kafka集群之间制作镜像数据的工具。该工具主要动作就是从源集群中消费并生产到目标群集。

一个集群可以启动多个MirrorMaker配置到多个集群

 

运行 MirrorMaker方法

kafka-mirror-maker.sh --consumer.config ./consumer.properties --producer.config ./producer.properties --num.streams 8 --whitelist ".*"

实际操作一遍

consumer.properties文件

root@DESKTOP-I0EG1MJ:~/kafka-client/mirror-maker# cat consumer.properties
bootstrap.servers=192.168.3.194:9092
group.id=mirrormaker
auto.offset.reset=earliest

producer.properties文件

root@DESKTOP-I0EG1MJ:~/kafka-client/mirror-maker# cat producer.properties
bootstrap.servers=192.168.2.123:9092

 运行mirror-maker

root@DESKTOP-I0EG1MJ:~/kafka-client/mirror-maker# cat run-mirrormaker.sh
kafka-mirror-maker.sh --consumer.config ./consumer.properties --producer.config ./producer.properties --offset.commit.interval.ms 5000 --num.streams 2 --whitelist "test0428"

参数说明：

--consumer.config 消费者的配置文件（要消费的集群）

--producer.config 指定生产配置文件（要发送到的目标集群）

--whitelist 要同步的topic白名单，可以匹配正则，也可以指定具体topic

--offset.commit.interval.ms 消费端提交offset时间间隔

--num.streams  MirrorMaker 要创建多少个 KafkaConsumer 实例

更多可以使用 --help查看

root@DESKTOP-I0EG1MJ:~/kafka-client/mirror-maker# kafka-mirror-maker.sh --help
This tool helps to continuously copy data between two Kafka clusters.
Option                                   Description
------                                   -----------
--abort.on.send.failure <String: Stop    Configure the mirror maker to exit onthe entire mirror maker when a send      a failed send. (default: true)failure occurs>
--consumer.config <String: config file>  Embedded consumer config for consumingfrom the source cluster.
--consumer.rebalance.listener <String:   The consumer rebalance listener to useA custom rebalance listener of type      for mirror maker consumer.ConsumerRebalanceListener>
--help                                   Print usage information.
--message.handler <String: A custom      Message handler which will processmessage handler of type                  every record in-between consumer andMirrorMakerMessageHandler>               producer.
--message.handler.args <String:          Arguments used by custom messageArguments passed to message handler      handler for mirror maker.constructor.>
--new.consumer                           DEPRECATED Use new consumer in mirrormaker (this is the default so thisoption will be removed in a futureversion).
--num.streams <Integer: Number of        Number of consumption streams.threads>                                 (default: 1)
--offset.commit.interval.ms <Integer:    Offset commit interval in ms.offset commit interval in                (default: 60000)millisecond>
--producer.config <String: config file>  Embedded producer config.
--rebalance.listener.args <String:       Arguments used by custom rebalanceArguments passed to custom rebalance     listener for mirror maker consumer.listener constructor as a string.>
--version                                Display Kafka version.
--whitelist <String: Java regex          Whitelist of topics to mirror.(String)>

 

运行后会提示这么一句话

WARNING: The default partition assignment strategy of the mirror maker will change from 'range' to 'roundrobin' in an upcoming release (so that better load balancing can be achieved). If you prefer to make this switch in advance of that release add the following to the corresponding config: 'partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor'

 主要是说是MirrorMaker使用的消费策略是 Range ，以后可能改成 “轮训策略” ，我们可以手动指定“轮询策略”

在Consuemr.properties中设置即可

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

 

 

此时我们已经运行成功，直接去查看是否可以消费成功即可。

注意：同步前最后将目标集群的Topic创建好，否则会使用Broker默认配置