Kafka——MirrorMaker详解

目录

MirrorMaker

1、概要介绍

对于Kafka企业级用户而言，一个常见的痛点就是跨机房或跨数据中心(data center,DC)的数据传输。大型企业通常在多个数据中心部署Kafka集群。.这里的数据中心可能是企业拥有的自建机房，也可能是公有云厂商的不同机房。在多个机房部署Kafk集群的优势如下。

• 实现灾备。
• 较近的地理位置可缩短延时以及用户响应时间。
• 实现负载均衡，即每个数据中心上的集群可能只保存部分数据集合。
• 区别隔离不同优先级的数据处理。

尽管部署方式各种各样，但跨机房部署方案有一个共同的特点：数据需要能够从一个Kafka集群被拷贝到另一个集群，而且还必须支持双向拷贝：某次传输的源集群(source cluster)可能是下次传输的目标集群(target cluster)。

为了实现这样的需求，Kafka默认提供了一个工具MirrorMaker,用来帮助用户实现数据在两个Kafka集群间的拷贝。就具体实现而言，MirrorMaker仅仅是一个consumer+producer的混合体。对于源集群而言，它是一个consumer;而对于目标集群而言，它又是一个producer。
MirrorMaker读取源集群指定topic的数据，然后写入目标集群中的同名topic下。用户可以运行多个MirrorMaker实例增加整体数据拷贝的吞吐量，同时还提升了容错性。毕竟当一个实例崩溃后，其他实例能够自动地承担起它的负载。

在实际生产环境中，源集群和目标集群是完全独立的两套环境：它们上的toc可能设置了不同的分区数且有不同的位移值。基于这个原因，MirrorMaker工具并不能完美地实现容错性一因为consumer的位移值可能是不同的。不过，MirrorMaker依然会保存并使用消息的key来执行分区任务。

2、主要参数

MirrorMaker的脚本名是kafka-mirror–maker…sh(bat),位于Kafka安装目录的bin子目录下
(Windows平台下则位于Kafka安装目录的bin/windows子目录下)。
参数名参数含义whitelist指定一个正则表达式，指定拷贝源集群中的哪些topic。比如 a|b 表示拷贝源集群上连个topic的数据a和b。注意，当使用新版本consumer时必须指定该参数。为了方便使用，可以将|替换为,blacklist指定一个正则表达式，屏蔽指定topic的拷贝。注意，该参数只使用于老版本consumerabort.on.send.failure若设置为true，当发送失败时则关闭MirrorMakerconsumer.config指定MirrorMaker下consumer的属性文件。至少指定bootstrap.server和group.idproducer.config指定MirrorMaker下producer的属性文件consumer.rebalance.listener指定MirrorMaker使用的consumer rebalance监听器rebalance.listener.args指定MirrorMaker使用的consumer rebalance监听器的参数，与consumer.rebalance.listener一同使用message.handler指定消息处理器类。消息处理器在consumer获取消息与producer发送消息之间调用message.handler.args指定消息处理器类的参数，与message.handler一同使用num.streams指定MirrorMaker线程数。默认是1offset.conmmit.interval.ms指定MirrorMaker位移提交间隔，默认值为1分钟help打印帮助信息
在实际使用中，经常被使用的参数是whitelist、.consumer.config和producer…config。如果
要实现某些特定的拷贝逻辑（比如拷贝指定分区数据或有选择性地拷贝数据等），那么就需要实现特定的消息处理器并使用message.handler进行指定。

一个典型的命令执行如下。该命令读取由consumer.properties文件指定的源Kafka集群，
去读取名为topicA和topicB主题的消息，并写入到由producer.properties文件指定的目标Kafka集群。

bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB

3、实例

下面我们结合一个具体的实例来看看如何应用MirrorMaker工具在集群间进行数据拷贝。

在本例中我们搭建2个测试Kafka单点集群，分别用k1、k2表示。两套环境都是单节点的Kafka集群，端口号分别是9092、9093。本例的目标是在k1上生产消息，然后使用MirrorMaker将消息拷贝到k2上，最后在k2上运行consumer以验证消息是否被成功拷贝。

分别设置2套环境，使用不同的Zookeeper chroot：

zookeeper.connect=localhost:2181/k1
zookeeper.connect=localhost:2181/k2

分别启动k1、k2

./kafka-server-start.sh -daemon../config/server.properties

./kafka-server-start.sh -daemon../config/server2.properties

在k1上创建一个测试topic

./kafka-topics.sh --bootstrap-server localhost:9092 --create--topic test-topic --partitions1 --replication-factor 1

创建consumer.config文件

bootstrap.servers=localhost:9092
client.id=test.k1Andk2
group.id=test.k1Andk2
partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

创建producer.config文件

bootstrap.servers=localhost:9093
client.id=test.k1Andk2

启动MirrorMaker

./kafka-mirror-maker.sh --consumer.config../config/consumer.properties --producer.config../config/producer.properties --whitelist test-topic

在k1上是使用console producer模拟生产消息

./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
one
two
three
four

在k2上使用console consumer验证消息是否拷贝成功：

./kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test-topic --from-beginning

上述输出表明k1的消息经由MirrorMaker成功拷贝到k2集群。