Kafka跨集群迁移方案MirrorMaker原理、使用以及性能调优实践 - CSDN博客

标签： | 发表时间：2018-07-22 19:55 | 作者：
出处：https://blog.csdn.net

序言

Kakfa MirrorMaker是Kafka 官方提供的跨数据中心的流数据同步方案。其实现原理，其实就是通过从Source Cluster消费消息然后将消息生产到Target Cluster，即普通的消息生产和消费。用户只要通过简单的consumer配置和producer配置，然后启动Mirror，就可以实现准实时的数据同步。

1. Kafka MirrorMaker基本特性

Kafka Mirror的基本特性有：

  1. 在Target Cluster没有对应的Topic的时候，Kafka MirrorMaker会自动为我们在Target Cluster上创建一个一模一样(Topic Name、分区数量、副本数量)一模一样的topic。如果Target Cluster存在相同的Topic则不进行创建，并且，MirrorMaker运行Source Cluster和Target Cluster的Topic的分区数量和副本数量不同。
  2. 同我们使用Kafka API创建KafkaConsumer一样，Kafka MirrorMaker允许我们指定多个Topic。比如，TopicA|TopicB|TopicC。在这里，|其实是正则匹配符，MirrorMaker也兼容使用逗号进行分隔。
  3. 多线程支持。MirrorMaker会在每一个线程上创建一个Consumer对象，如果性能允许，建议多创建一些线程
  4. 多进程任意横向扩展，前提是这些进程的consumerGroup相同。无论是多进程还是多线程，都是由Kafka ConsumerGroup的设计带来的任意横向扩展性，具体的分区分派，即具体的TopicPartition会分派给Group中的哪个Topic负责，是Kafka自动完成的，Consumer无需关心。 
    我们使用Kafka MirrorMaker完成远程的AWS(Source Cluster)上的Kafka信息同步到公司的计算集群(Target Cluster)。由于我们的大数据集群只有一个统一的出口IP，外网访问我们的内网服务器必须通过nginx转发，因此为了降低复杂度，决定使用“拉”模式而不是“推”模式，即，Kafka MirrorMaker部署在我们内网集群(Target Cluster)，它负责从远程的Source Cluster(AWS)的Kafka 上拉取数据，然后生产到本地的Kafka。

Kafka MirrorMaker的官方文档一直没有更新，因此新版Kafka为MirrorMaker增加的一些参数、特性等在文档上往往找不到，需要看Kafka MirrorMaker的源码。Kafka MirrorMaker的主类位于 kafka.tools.MirrorMaker，尤其是一些参数的解析逻辑和主要的执行流程，会比较有助于我们理解、调试和优化Kafka MirrorMaker。

2. 新旧Consumer API的使用问题

从Kafka 0.9版本开始引入了new consumer API。相比于普通的old consumer api，new Conumser API有以下主要改变：

  • 统一了旧版本的High-Level和Low-Level Consumer API；
  • new consumer API消除了对zookeeper的依赖，修改了ConsumerGroup的管理等等协议；
  • new Consumer API完全使用Java实现，不再依赖Scala环境；
  • 更好了安全认证只在new consumer中实现，在old consumer中没有。

Kakfa MirrorMaker同时提供了对新旧版本的Consumer API的支持。 
默认是旧版API，当添加 --new.consumer参数，MirrorMaker将使用新的Consumer API进行消息消费：

   // Create consumers
      val mirrorMakerConsumers = if (!useNewConsumer) {//如果用户没有配置使用new consumer，则使用旧的consumer
        val customRebalanceListener = {
          val customRebalanceListenerClass = options.valueOf(consumerRebalanceListenerOpt)
          if (customRebalanceListenerClass != null) {
            val rebalanceListenerArgs = options.valueOf(rebalanceListenerArgsOpt)
            if (rebalanceListenerArgs != null) {
              Some(CoreUtils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass, rebalanceListenerArgs))
            } else {
              Some(CoreUtils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass))
            }
          } else {
            None
          }
        }

        if (customRebalanceListener.exists(!_.isInstanceOf[ConsumerRebalanceListener]))
          throw new IllegalArgumentException("The rebalance listener should be an instance of kafka.consumer.ConsumerRebalanceListener")
        createOldConsumers(//创建旧的consumer
          numStreams,
          options.valueOf(consumerConfigOpt),
          customRebalanceListener,
          Option(options.valueOf(whitelistOpt)),
          Option(options.valueOf(blacklistOpt)))
      } else {//用户指定使用new  consumer
        val customRebalanceListener = {
          val customRebalanceListenerClass = options.valueOf(consumerRebalanceListenerOpt)
          if (customRebalanceListenerClass != null) {
            val rebalanceListenerArgs = options.valueOf(rebalanceListenerArgsOpt)
            if (rebalanceListenerArgs != null) {
                Some(CoreUtils.createObject[org.apache.kafka.clients.consumer.ConsumerRebalanceListener](customRebalanceListenerClass, rebalanceListenerArgs))
            } else {
                Some(CoreUtils.createObject[org.apache.kafka.clients.consumer.ConsumerRebalanceListener](customRebalanceListenerClass))
            }
          } else {
            None
          }
        }
        if (customRebalanceListener.exists(!_.isInstanceOf[org.apache.kafka.clients.consumer.ConsumerRebalanceListener]))
          throw new IllegalArgumentException("The rebalance listener should be an instance of" +
            "org.apache.kafka.clients.consumer.ConsumerRebalanceListner")
        createNewConsumers(//创建new consumer
          numStreams,
          options.valueOf(consumerConfigOpt),
          customRebalanceListener,
          Option(options.valueOf(whitelistOpt)))
      }   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11
    •     
  • 12
    •     
  • 13
    •     
  • 14
    •     
  • 15
    •     
  • 16
    •     
  • 17
    •     
  • 18
    •     
  • 19
    •     
  • 20
    •     
  • 21
    •     
  • 22
    •     
  • 23
    •     
  • 24
    •     
  • 25
    •     
  • 26
    •     
  • 27
    •     
  • 28
    •     
  • 29
    •     
  • 30
    •     
  • 31
    •     
  • 32
    •     
  • 33
    •     
  • 34
    •     
  • 35
    •     
  • 36
    •     
  • 37
    •     
  • 38
    •     
  • 39
    •     
  • 40
    •     
  • 41
    •     
  • 42
    •     
  • 43
    •     
  • 44
    •     
  • 45
    •     
  • 46
    •     
  • 47

这是我启动Kakfa MirrorMaker 的命令：

   nohup ./bin/kafka-mirror-maker.sh --new.consumer  --consumer.config config/mirror-consumer.properties  --num.streams 40 --producer.config config/mirror-producer.properties  --whitelist 'ABTestMsg|AppColdStartMsg|BackPayMsg|WebMsg|GoldOpenMsg|BoCaiMsg' &   
        
  • 1

mirror-consumer.properties配置文件如下：

   #新版consumer摈弃了对zookeeper的依赖，使用bootstrap.servers告诉consumer kafka server的位置
bootstrap.servers=ip-188-33-33-31.eu-central-1.compute.internal:9092,ip-188-33-33-32.eu-central-1.compute.internal:9092,ip-188-33-33-33.eu-central-1.compute.internal:9092

#如果使用旧版Consumer，则使用zookeeper.connect
#zookeeper.connect=ip-188-33-33-31.eu-central-1.compute.internal:2181,ip-188-33-33-32.eu-central-1.compute.internal:2181,ip-188-33-33-33.eu-central-1.compute.internal:2181
1.compute.internal:2181
#change the default 40000 to 50000
request.timeout.ms=50000

#hange default heartbeat interval from 3000 to 15000
heartbeat.interval.ms=30000

#change default session timeout from 30000 to 40000
session.timeout.ms=40000
#consumer group id
group.id=africaBetMirrorGroupTest
partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor
#restrict the max poll records from 2147483647 to 200000
max.poll.records=20000
#set receive buffer from default 64kB to 512kb
receive.buffer.bytes=524288

#set max amount of data per partition to override default 1048576
max.partition.fetch.bytes=5248576
#consumer timeout
#consumer.timeout.ms=5000   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11
    •     
  • 12
    •     
  • 13
    •     
  • 14
    •     
  • 15
    •     
  • 16
    •     
  • 17
    •     
  • 18
    •     
  • 19
    •     
  • 20
    •     
  • 21
    •     
  • 22
    •     
  • 23
    •     
  • 24
    •     
  • 25
    •     
  • 26

mirror-producer.properties的配置文件如下：

   bootstrap.servers=10.120.241.146:9092,10.120.241.82:9092,10.120.241.110:9092

# name of the partitioner class for partitioning events; default partition spreads data randomly
#partitioner.class=

# specifies whether the messages are sent asynchronously (async) or synchronously (sync)
producer.type=sync

# specify the compression codec for all data generated: none, gzip, snappy, lz4.
# the old config values work as well: 0, 1, 2, 3 for none, gzip, snappy, lz4, respectively
compression.codec=none
# message encoder
serializer.class=kafka.serializer.DefaultEncoder   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11
    •     
  • 12
    •     
  • 13

同时，我使用 kafka-consumer-groups.sh循环监控消费延迟：

   bin/kafka-consumer-groups.sh  --bootstrap-server ip-188-33-33-31.eu-central-1.compute.internal:9092,ip-188-33-33-32.eu-central-1.compute.internal:9092,ip-188-33-33-33.eu-central-1.compute.internal:9092 --describe --group africaBetMirrorGroupTest  --new-consumer   
        
  • 1

当我们使用new KafkaConsumer进行消息消费，要想通过kafka-consumer-groups.sh获取整个group的offset、lag延迟信息，也必须加上–new-consumer，告知kafka-consumer-groups.sh，这个group的消费者使用的是new kafka consumer，即group中所有consumer的信息保存在了Kafka上的一个名字叫做 __consumer_offsets的特殊topic上，而不是保存在zookeeper上。我在使用kafka-consumer-groups.sh的时候就不知道还需要添加 --new-consumer，结果我启动了MirrorMaker以后，感觉消息在消费，但是就是在zookeeper的/consumer/ids/上找不到group的任何信息。后来在stack overflow上问了别人才知道。

3. 负载不均衡原因诊断以及问题解决

在我的另外一篇博客 《Kafka为Consumer分派分区：RangeAssignor和RoundRobinAssignor》中，介绍了Kafka内置的分区分派策略：RangeAssignor和RoundRobinAssignor。由于RangeAssignor是早期版本的Kafka的唯一的分区分派策略，因此，默认不配置的情况下，Kafka使用RangeAssignor进行分区分派，但是，在MirrorMaker的使用场景下，RoundRobinAssignor更有利于均匀的分区分派。甚至在 KAFKA-3831中有人建议直接将MirrorMaker的默认分区分派策略改为RoundRobinAssignor。那么，它们到底有什么区别呢？我们先来看两种策略下的分区分派结果。在我的实验场景下，有6个topic： ABTestMsg|AppColdStartMsg|BackPayMsg|WebMsg|GoldOpenMsg|BoCaiMsg，每个topic有两个分区。由于MirrorMaker所在的服务器性能良好，我设置 --num.streams 40，即单台MirrorMaker会用40个线程，创建40个独立的Consumer进行消息消费，两个MirrorMaker加起来80个线程，80个并行Consumer。由于总共只有 6 * 2=12个TopicPartition，因此最多也只有12个Consumer会被分派到分区，其余Consumer空闲。 
我们来看基于RangeAssignor分派策略，运行kafka-consumer-groups.sh观察到的分区分派的结果：

   TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
ABTestMsg                      0          780000          820038          49938      africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-0
ABTestMsg                      1          774988          820038          55000      africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-1
AppColdStartMsg                0          774000          820039          55938      africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-0
AppColdStartMsg                1          774100          820045          56038      africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-1
BackPayMsg                     0          780000          820038          49938      africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-0
BackPayMsg                     1          774988          820038          55000      africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-1
WebMsg                         0          774000          820039          55938      africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-0
WebMsg                         1          774100          820045          56038      africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-1
GoldOpenMsg                    0          780000          820038          49938      africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-0
GoldOpenMsg                    1          774988          820038          55000      africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-1
BoCaiMsg                       0          774000          820039          55938      africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-0
BoCaiMsg                       1          774100          820045          56038      africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-1
-                              -          -               -               -          africaBetMirrorGroupTest-6-ae373364-2ae2-42b8-8a74-683557e315bf/114.113.198.126               africaBetMirrorGroupTest-6
-                              -          -               -               -          africaBetMirrorGroupTest-9-0e346b46-1a2c-46a2-a2da-d977402f5c5d/114.113.198.126               africaBetMirrorGroupTest-9
-                              -          -               -               -          africaBetMirrorGroupTest-7-f0ae9f31-33e6-4ddd-beac-236fb7cf20d5/114.113.198.126               africaBetMirrorGroupTest-7
-                              -          -               -               -          africaBetMirrorGroupTest-7-e2a9e905-57c1-40a6-a7f3-4aefd4f1a30a/114.113.198.126               africaBetMirrorGroupTest-7
-                              -          -               -               -          africaBetMirrorGroupTest-8-480a2ef5-907c-48ed-be1f-33450903ec72/114.113.198.126               africaBetMirrorGroupTest-8
-                              -          -               -               -          africaBetMirrorGroupTest-8-4206bc08-58a5-488a-b756-672fb4eee6e0/114.113.198.126               africaBetMirrorGroupTest-8
.....后续更多空闲consumer省略不显示   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11
    •     
  • 12
    •     
  • 13
    •     
  • 14
    •     
  • 15
    •     
  • 16
    •     
  • 17
    •     
  • 18
    •     
  • 19
    •     
  • 20

当没有在 mirror-consumer.properties 中配置分区分派策略，即使用默认的RangeAssignor的时候，我们发现，尽管我们每一个MirrorMaker有40个Consumer，整个Group中有80个Consumer，但是，一共 6 * 2 = 12个TopicPartition竟然全部聚集在2-3个Consumer上，显然，这完全浪费了并行特性，被分配到一个consumer上的多个TopicPartition只能串行消费。

因此，通过 partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor显式指定分区分派策略为RoundRobinAssignor，重启MirrorMaker，重新通过 kafka-consumer-groups.sh 命令观察分区分派和消费延迟结果：

   TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
ABTestMsg                      0          819079          820038          959        africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-1
ABTestMsg                      1          818373          820038          1665       africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-5
AppColdStartMsg                0          818700          818907          1338       africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-20
AppColdStartMsg                1          818901          820045          1132       africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-18
BackPayMsg                     0          819032          820038          959        africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-5
BackPayMsg                     1          818343          820038          1638       africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-8
WebMsg                         0          818710          818907          1328       africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-7
WebMsg                         1          818921          820045          1134       africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-9
GoldOpenMsg                    0          819032          820038          959        africaBetMirrorGroupTest-4-cf330e66-1319-4925-9605-46545df13453/114.113.198.126               africaBetMirrorGroupTest-12
GoldOpenMsg                    1          818343          820038          1638       africaBetMirrorGroupTest-19-c77523e7-7b87-472b-9a26-cd902888944d/114.113.198.126              africaBetMirrorGroupTest-14
BoCaiMsg                       0          818710          818907          1322       africaBetMirrorGroupTest-19-674d8ad4-39d2-40cc-ae97-f4be9c1bb154/114.113.198.126              africaBetMirrorGroupTest-14
BoCaiMsg                       1          818921          820045          1189       africaBetMirrorGroupTest-15-91c67bf8-0c1c-42ac-97f0-5369794c2d1b/114.113.198.126              africaBetMirrorGroupTest-117
-                              -          -               -               -          africaBetMirrorGroupTest-6-ae373364-2ae2-42b8-8a74-683557e315bf/114.113.198.126               africaBetMirrorGroupTest-6
-                              -          -               -               -          africaBetMirrorGroupTest-9-0e346b46-1a2c-46a2-a2da-d977402f5c5d/114.113.198.126               africaBetMirrorGroupTest-9
-                              -          -               -               -          africaBetMirrorGroupTest-7-f0ae9f31-33e6-4ddd-beac-236fb7cf20d5/114.113.198.126               africaBetMirrorGroupTest-7
-                              -          -               -               -          africaBetMirrorGroupTest-7-e2a9e905-57c1-40a6-a7f3-4aefd4f1a30a/114.113.198.126               africaBetMirrorGroupTest-7
-                              -          -               -               -          africaBetMirrorGroupTest-8-480a2ef5-907c-48ed-be1f-33450903ec72/114.113.198.126               africaBetMirrorGroupTest-8
-                              -          -               -               -          africaBetMirrorGroupTest-8-4206bc08-58a5-488a-b756-672fb4eee6e0/114.113.198.126               africaBetMirrorGroupTest-8
.....后续更多空闲consumer省略不显示   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11
    •     
  • 12
    •     
  • 13
    •     
  • 14
    •     
  • 15
    •     
  • 16
    •     
  • 17
    •     
  • 18
    •     
  • 19
    •     
  • 20

对比RangeAssingor，消息延迟明显减轻，而且，12个TopicPartition被均匀分配到了不同的consumer上，即单个Consumer只负责一个TopicPartition的消息消费，不同的TopicPartition之间实现了完全并行。 
之所以出现以上不同，原因在于两个分区分派方式的策略不同：

  • RangeAssingor：先对所有Consumer进行排序，然后对Topic逐个进行分区分派。用以上Topic作为例子：
   对所有的Consumer进行排序，排序后的结果为Consumer-0,Consumer-1,Consumer-2 ....Consumer-79
对ABTestMsg进行分区分派：
ABTestMsg-0分配给Consumer-0
ABTestMsg-1分配各Consumer-1

对AppColdStartMsg进行分区分派：
AppColdStartMsg-0分配各Consumer-0
AppColdStartMsg-1分配各Consumer-1

#后续TopicParition的分派以此类推
   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9
    •     
  • 10
    •     
  • 11

可见，RangeAssingor 会导致多个TopicPartition被分派在少量分区上面。 
- RoundRobinAssignor：与RangeAssignor最大的区别，是不再逐个Topic进行分区分派，而是先将Group中的所有TopicPartition平铺展开，再一次性对他们进行一轮分区分派。

将Group中的所有TopicPartition展开，展开结果为：

   ABTestMsg-0,ABTestMsg-1,AppColdStartMsg-0,AppColdStartMsg-1,BackPayMsg-0,BackPayMsg-1,WebMsg-0，WebMsg-1，GoldOpenMsg-0，GoldOpenMsg-1，BoCaiMsg-0，BoCaiMsg-1   
        
  • 1

对所有的Consumer进行排序，排序后的结果为 Consumer-0， Consumer-1， Consumer-2 ， Consumer-79。

开始讲平铺的TopicPartition进行分区分派

   ABTestMsg-0分配给Consumer-0
ABTestMsg-1分配给Consumer-1
AppColdStartMsg-0分配给Consumer-2
AppColdStartMsg-1分配给Consumer-3
BackPayMsg-0分配给Consumer-4
BackPayMsg-1分配给Consumer-5


#后续TopicParition的分派以此类推   
        
  • 1
    •     
  • 2
    •     
  • 3
    •     
  • 4
    •     
  • 5
    •     
  • 6
    •     
  • 7
    •     
  • 8
    •     
  • 9

由此可见，RoundRobinAssignor平铺式的分区分派算法是让我们的Kafka MirrorMaker能够无重叠地将TopicParition分派给Consumer的原因。

4. 本身网络带宽限制问题

网络带宽本身也会限制Kafka Mirror的吞吐量。进行压测的时候，我分别在我们的在线环境和测试环境分别运行Kafka MirrorMaker，均选择两台服务器运行MirrorMaker，但是在线环境是实体机环境，单台机器通过SCP方式拷贝Source Cluster上的大文件，平均吞吐量是600KB-1.5MB之间，但是测试环境的机器是同一个host主机上的多台虚拟机，SCP吞吐量是100KB以下。经过测试，测试环境消息积压会逐渐增多，在线环境持续积压，但是积压一直保持稳定。这种稳定积压是由于每次 poll()的间隙新产生的消息量，属于正常现象。

5. 适当配置单次poll的消息总量和单次poll()的消息大小

通过Kafka MirrorMaker运行时指定的consumer配置文件(在我的环境中为 $MIRROR_HOME/config/mirror-consumer.properties)来配置consumer。其中，通过以下配置，可以控制单次 poll()的消息体量（数量和总体大小） 
max.poll.records：单次 poll()操作最多消费的消息总量，这里说的poll是单个consumer而言的。无论过大过小，都会发生问题：

  • 如果设置得过小，则消息传输率降低，大量的头信息会占用较大的网络带宽；-
  • 如果设置得过大，则会产生一个非常难以判断原因同时又会影响整个group中所有消息的消费的重要问题：rebalance。看过kafka代码的话可以看到，每次poll()请求都会顺带向远程server发送心跳信息，远程GroupCoordinator会根据这个心跳信息判断consumer的活性。如果超过指定时间( heartbeat.interval.ms)没有收到对应Consumer的心跳，则GroupCoordinator会判定这个Server已经挂掉，因此将这个Consumer负责的partition分派给其它Consumer，即触发rebalance。rebalance操作的影响范围是整个Group，即Group中所有的Consumer全部暂停消费直到Rebalance完成。而且，TopicPartition越长，这个过程会越长。其实，一个正常消费的环境，应该是任何时候都不应该发生rebalance的(一个新的Consumer的正常加入也会引起Rebalance，这种情况除外)。虽然Kafka本身是非常稳定的，但是还是应该尽量避免rebalance的发生。在某些极端情况下触发一些bug，rebalance可能永远停不下来了。。。如果单次 max.poll.records消费太多消息，这些消息produce到Target Cluster的时间可能会较长，从而可能触发Rebalance。

6. 恶劣网络环境下增加超时时间配置

在不稳定的网络环境下，应该增加部分超时时间配置，如 request.timeout.ms或者 session.timeout.ms，一方面可以避免频繁的超时导致大量不必要的重试操作，同时，通过增加如上文所讲，通过增加 heartbeat.interval.ms时间，可以避免不必要的rebalance操作。当然，在网络环境良好的情况下，上述配置可以适当减小以增加Kafka Server对MirrorMaker出现异常情况下的更加及时的响应。

总之，Kafka MirrorMaker作为跨数据中心的Kafka数据同步方案，绝对无法允许数据丢失以及数据的传输速度低于生产速度导致数据越积累越多。因此，唯有进行充分的压测和精准的性能调优，才能综合网络环境、服务器性能，将Kafka MirrorMaker的性能发挥到最大。

相关 [kafka 集群 mirrormaker] 推荐：

Kafka跨集群迁移方案MirrorMaker原理、使用以及性能调优实践 - CSDN博客

- -
Kakfa MirrorMaker是Kafka 官方提供的跨数据中心的流数据同步方案. 其实现原理，其实就是通过从Source Cluster消费消息然后将消息生产到Target Cluster，即普通的消息生产和消费. 用户只要通过简单的consumer配置和producer配置，然后启动Mirror，就可以实现准实时的数据同步.

Kafka MirrorMaker实践 - (a != b) ? b : a - ITeye博客

- -
最近准备使用Kafka Mirrormaker做两个数据中心的数据同步，以下是一些要点：. mirrormaker必须提供一个或多个consumer配置，一个producer配置，一个whitelist或一个blacklist（支持java正则表达式）. 启动多个mirrormaker进程，单个进程启动多个consuemr streams， 可以提高吞吐量和提高性能.

kafka集群安装

- - 互联网 - ITeye博客
kafka是LinkedIn开发并开源的一个分布式MQ系统，现在是Apache的一个孵化项目. 在它的主页描述kafka为一个高吞吐量的分布式（能将消息分散到不同的节点上）MQ. 在这片博文中，作者简单提到了开发kafka而不选择已有MQ系统的原因. Kafka仅仅由7000行Scala编写，据了解，Kafka每秒可以生产约25万消息（50 MB），每秒处理55万消息（110 MB）.

Kafka跨数据中心迁移方案MirrorMaker使用及性能调优实践 | 网易乐得技术团队

- -
Kakfa MirrorMaker是Kafka 官方提供的跨数据中心的流数据同步方案. 其实现原理，其实就是通过从Source Cluster消费消息然后将消息生产到Target Cluster，即普通的消息生产和消费. 用户只要通过简单的consumer配置和producer配置，然后启动Mirror，就可以实现准实时的数据同步.

kafka集群操作命令

- - 开源软件 - ITeye博客
默认Kafka会使用ZooKeeper默认的/路径，这样有关Kafka的ZooKeeper配置就会散落在根路径下面，如果 你有其他的应用也在使用ZooKeeper集群，查看ZooKeeper中数据可能会不直观，所以强烈建议指定一个chroot路径，直接在 zookeeper.connect配置项中指定.

【译】调优Apache Kafka集群 - huxihx - 博客园

- -
　　今天带来一篇译文“调优Apache Kafka集群”，里面有一些观点并无太多新颖之处，但总结得还算详细. 该文从四个不同的目标出发给出了各自不同的参数配置，值得大家一读~ 原文地址请参考：https://www.confluent.io/blog/optimizing-apache-kafka-deployment/.

Kafka系列（八）跨集群数据镜像

- - Dengshenyu
本系列文章为对《Kafka：The Definitive Guide》的学习整理，希望能够帮助到大家. 在之前系列文章中，我们讨论了一个Kafka集群的搭建、维护和使用，而在实际情况中我们往往拥有多个Kafka集群，而且这些Kafka集群很可能是相互隔离的. 一般来说，这些集群之间不需要进行数据交流，但如果在某些情况下这些集群之间存在数据依赖，那么我们可能需要持续的将数据从一个集群复制到另一个集群.

Kafka的集群部署实践及运维相关

- - DockOne.io
上一篇 Kafka 的文章《 大白话带你认识Kafka》中我们应该已经了解了一些关于基础角色和集群架构相关的问题，这时候我们应该很想了解一下如何构建生产中的Kafka集群或者一些相关的运维工具，所以就应运而生了下文. 假设每天集群需要承载10亿数据. 一天24小时，晚上12点到凌晨8点几乎没多少数据.

kafka集群broker节点扩容方案_watermelonbig的专栏-CSDN博客

- -
在用kafka集群有3个节点，即host1, host2, host3，现需要将broker节点扩容至6个，以提供更高的数据处理能力. 一、上架物理服务器用于提供更多的资源. 新扩容3个broker节点，host4,host5,host6. 二、在三台新增节点上部署kafka应用程序. 三、重新分布原有的topic分区.

kafka监控之kafka-run-class.sh

- - 开源软件 - ITeye博客
kafka自带了很多工具类，在源码kafka.tools里可以看到：. 这些类该如何使用呢，kafka的设计者早就为我们考虑到了，在${KAFKA_HOME}/bin下，有很多的脚本，其中有一个kafka-run-class.sh，通过这个脚本，可以调用其中的tools的部分功能，如调用kafka.tools里的ConsumerOffsetChecker.scala,.