前言

异地灾备可以防止一个机房出现灾难而导致整个系统无法提供服务的问题。
异地双活在异地灾备的基础上减少了数据的冗余，并且提高了对数据双向同步及同步实时性的要求。
Kafka作为在双活中扮演了很重要的角色，一方面是两个区域的正常业务的消息数据分发、另一方面则是elasticsearch等中间件集群的双活方案有时需要依赖kafka的双活。那么在保证实时性的前提下做好kafka的双活呢？下面博主抛砖引玉给出自己实现的基于Kafka自带组件Mirror-Maker的两种方案，读者可以根据自己需要进行改进。

一、Kafka-Mirror-Maker

MirrorMaker是Kafka官方提供的用来做跨机房同步的组件。在kafka的安装目录的bin目录下有一个kafka-mirror-maker.sh文件就是MirrorMaker的入口。

image.png

• 工作原理：

1、MirrorMaker本质上既是一个生产者又是一个消费者
2、MirrorMaker从一个机房的某个主题消费消息，再把消费到的消息生产到另一个机房的相同的主题下。

• 如何使用：

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

参数解释：
--producer.config：mirrormaker中的生产者，与我们常规的生产者配置是一样的，我们可以参照生产者配置文档进行参数配置
--consumer.config：mirrormaker中的消费者，与我们常规的消费者配置相同，可以参照消费者配置文档进行参数配置
--num.streams：mirrormaker中消费者线程的个数
--whitelist：mirrormaker中消费者指定消费的主题，.*表示所有主题，常规可以使用 | 分割开多个主题。
当然还有很多其他参数，我们可以执行bin/kafka-mirror-maker.sh --help来查看

• 疑问解答：

1、mirrormaker应该从哪个机房消费，向哪个机房生产。
答：应该从异地机房消费向本地机房生产。如果反过来，当网络异常的时候就会产生消费的消息没有及时得到转发。
2、mirrormaker怎么配置生产者和消费者？
答：参照这个链接里的配置 就可以了 https://www.orchome.com/511

方案一、各机房自主题

• 实现原理

1、假设有北方机房和南方机房需要实现双活
2、南方机房的kafka包含江苏、广东、上海三个topic。北方机房包含黑龙江、甘肃、四川三个topic。每个机房的消费者只向自己机房的topic生产消息和消费消息。
3、启动两个Mirror-Maker分别负责从北向南同步数据和从南向北同步数据，保证两遍的机房都有六个省份的全量数据。
4、考虑异常情况，如果北方机房挂了或者南方机房挂了，那么另一个机房都会有全量数据可以提供服务。当然不排除小部分数据丢失，但灾难情况下是无法避免的。

• 存在问题及解决办法
  存在问题：

1、__consumer_offset主题无法双向同步，会造成消费死循环。这就导致切换机房的时候备用消费者只能从头消费来保证消息的不丢失，消息会大量重复。
2、即便是灾备集群也存在同样的问题，如果把consumer_offset同步到灾备集群，如果主集群的topic的partition和灾备集群的topic的partition的数目不一致，就会导致切换机房的时候备用消费者无法复用offset的情况发生，最终导致消息大量重复消费或者消息丢失。
对于问题1的解决办法
1、参照方案二，做数据的全量同步

对于问题2的解决办法：

解决办法1、根据消息时间戳来找到消费者组对应的offset，然后从offset开始消费，同时业务端需要做幂等处理。下面给出关键代码（对应的kafka版本的消息要有时间戳才行）。

        // 获取所有分区
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        // 把分区映射成TopicPartition
        List<TopicPartition> topicPartitionList = partitionInfos
                .stream()
                .map(info -> new TopicPartition(topic, info.partition()))
                .collect(Collectors.toList());
        // 给consumer分配这些分区
        consumer.assign(topicPartitionList);
        // 根据时间戳找到消费偏移量
        Map<TopicPartition, Long> partitionTimestampMap = topicPartitionList.stream()
                .collect(Collectors.toMap(tp -> tp, tp -> 时间戳));
        Map<TopicPartition, OffsetAndTimestamp> partitionOffsetMap = consumer.offsetsForTimes(partitionTimestampMap);
        // consumer从消费偏移量处开始消费
        partitionOffsetMap.forEach((tp, offsetAndTimestamp) -> consumer.seek(tp, offsetAndTimestamp.offset()));

解决方法2、consumer消费的时候把消费偏移量offset记录到底层数据库中，利用数据库的双活进行同步。切换机房的时候根据数据库中的offset进行消息的恢复。

各分区自主题.png

方案二、各机房同主题

• 背景

有的业务不能或者很难做topic主题拆分，这就要求两边机房的主题必须是相同的，并且消费者需要消费全量数据。但是原始的Mirror-Maker有一个特性，如果两遍机房的主题相同，那么互相搬运就会导致死循环消费消息。

• 实现原理：

1、假设有北方机房和南方机房需要实现双活
2、南方机房负责南方所有消息的分发，北方机房负责北方所有消息的分发
3、两方机房都要有全量数据并且主题不能拆分。
4、启动4个mirrormaker，分别负责北->北全、北->南全 、南->南全、南->北全的消息搬运

• 存在问题

1、业务层上需要过滤对方机房同步过来的消息，造成了逻辑耦合，对业务层的扩展性造成了限制和阻碍

image.png

• 疑问解答：

1、mirrormak只能把消息搬运到另一个机房的相同主题里，如何解决两个mirrormaker不会出现死循环搬运的问题？
答：MirrorMaker提供了一个--message.handler参数和一个--message.handler.args参数供开发者扩展。我们可以实现MessageHandler类，来对消息即将生产到的主题进行更改，更改方案github上有现成的，我们可以会直接使用即可：https://github.com/opencore/mirrormaker_topic_rename
当然你也可以自己实现其他的扩展，这里使用的时候需要注意kafka的版本。
2、如何使用上述的扩展源码呢？
答：很简单。
第一、mvn clean package 打成jar包
第二、export CLASSPATH=/home/sliebau/mmchangetopic-1.0-SNAPSHOT.jar 在你的机器上导入环境变量让kafka能够找到jar包
第三、--consumer.config consumer.properties --producer.config producer.properties --whitelist test_.* --message.handler com.opencore.RenameTopicHandler --message.handler.args `test_source,test_target;test_source2,test_target2` 启动mirrormaker的时候再加上两个参数，前面表示使用的类，后边表示类传入的参数，注意其中的格式。

二、实现高可伸缩

• 背景：

基于方案二。为了让mirrormaker服务能够随着数据量的增加而自由的伸缩，我们决定把mirrormaker服务化并且打包成docker。这样当数据量增大的时候我们可以通过增加mirror-maker的数量来承载。

• 服务版镜像地址：https://hub.docker.com/repository/docker/xiao5aha/mirror-maker

• 服务版镜像制作地址：https://github.com/KouLouYiMaSi/mirror-maker-service

• 如何使用这个镜像
  1、对于sasl_plaintext和scram加密，请使用xiao5aha/mirror-maker:v13_sasl版本

docker run -itd -e SOURCE=消费ip:9092 -e DESTINATION=生产到的ip:9092 -e GROUP_ID=消费者组ID -e USERNAME=用户名 -e PASSWORD=密码 -e FROM_TOPIC_TO_TOPIC="从哪个主题到哪个主题使用逗号隔开，多个主题对使用分号隔开" -e CONSUMER_COUNT=消费者数量  -e WHITE_LIST="消费者主题白名单" xiao5aha/mirror-maker:v13_sasl

2、对于没有加密的kafka请使用xiao5aha/mirror-maker:v9版本

docker run -itd -e SOURCE=ip:9092 -e DESTINATION=ip:9092 -e GROUP_ID=group -e MESSAGE_HANDLER=com.opencore.RenameTopicHandler -e MESSAGE_HANDLER_ARGS="nanfang-center,china" -e STREAM_COUNT=3  -e WHITE_LIST="wuxi-center" xiao5aha/mirror-maker:v9

• 如何制作镜像：

1、从上边的git地址拉下来源码
2、如果你想替换kafka版本可以自己从kafka官网下载一个（国内下载超级慢）替换，然后改下Dockerfile里边的版本就行了
3、如果你自己下载了kafka，用你的kafka版本替换我的kafka版本，注意是整个目录替换哦
4、如果需要修改mirrormaker的producer和consumer配置，打开scripts下边的start.sh，在下边图片部分添加或者删除或者修改。比如去掉其中的用户认证和权限认证。

image.png

5、全部修改完成之后docker build . docker tag 成自己的镜像就可以使用了

最后

转载请注明出处。