Kafka(分布式消息队列)集群迁移扩容

导语：Kafka：分布式消息队列，提供日志收集与消费。
1.版本介绍：
目前常用版本为kafka_2.9.2-0.8.1（scala-2.9.2为kafka官方推荐版本，此外还有2.8.2和2.10.2可以选择）
2.环境准备：
安装JDK6，目前使用版本为1.6.0_26，并配置JAVA_HOME
3.配置修改：
1）拷贝线上配置到本地kafka目录。
2）需要注意的是server.properties里broker和ip的指定，必须要唯一。
3）server.properties中log.dirs必须要手动指定。此配置项是kafka接收日志的存储目录(目前我们保存7天数据log.retention.hours=168)
4）zookeeper的指定可以是下面这种形式，手动定制zookeeper的目录，并且用逗号分隔ip
zookeeper.connect=192.168.0.76:2181,192.168.0.94:2181,192.168.0.99:2181/kafka-0.8.1 
5）默认分区和备份数分别为10和3（num.partitions=10 / default.replication.factor=3）当然这也可以在创建topic时手动指定。

4.部署：

目前kafka统一安装在/opt/apps_install/下。

JMX_PORT=9997 /opt/apps_install/kafka_2.9.2-0.8.1/bin/kafka-server-start.sh /opt/apps_install/kafka_2.9.2-0.8.1/config/server.properties >/dev/null 2>&1 &

部署新节点，然后通过命令启动kafka，JMX_PORT是为了打开kafka的可管理端口（默认不启用）。

5.命令：

常用增删改查命令：kafka基本命令，高级命令

1. 再简单介绍一些各节点的角色定义，如图显示：
   第一行中 ：Topic：tvadpv #此项为topic的名字。
                      PartitionCount:10 #此项为该topic的分区总数，分区数量越多，并发量越高。
                      ReplicationFactor:3 #此项为该topic的每个Partition的备份数量。
   第二行依次为：
                      Topic – topic名字
                      Partition – 分区号(因为咱们创建了10个partition，所以分区号是0~9)
                      Leader – 领导节点对应的broker id (因为咱们kafka机器公9个节点，所以broker id范围为0~8)
                      Replicas -- 该partition的备份所处于的broker id节点
                      Isr -- leader的备选节点(正常情况下，isr和replicas的broker id是对应的，如果其中缺少节点，说明对应kafka节点已经异常)

       

6.扩容：

1）先部署好新节点环境，并根据上文"配置修改"修改配置，然后启动集群，确保新节点为可用状态。

2）“高级命令“的介绍了扩容的基本方法：

1. 生成扩容使用的json文件：
   cat  topics-to-move.json
   

   {"topics": [{"topic": "tvadmclick"}],   //加粗部分是要手动指定的topic名称
   "version":1
   }

2. 通过上一步写好的json文件，使用kafka命令生成数据迁移配置(注意，如果手工移动数据，尽量不移动例如"replicas":[1,0,2]中的1号broker，也就是当前的leader，这样避免leader的重新选举，造成集群额外负担)

    bin/kafka-reassign-partitions.sh --topics-to-move-json-file topics-to-move.json --zookeeper 192.168.0.99:2181/kafka-0.8.1 --broker-list "0,1,2,3,4,5,6,7,8" --generate   //加粗部分是指定数据迁移到那些broker。命令执行完成后会生成类似下面的数据：

         
   

1. 将第一部分保存留作回退备份(即Proposed partition reassignment configuration上面的json串)，下面json串为扩容将要使用的到的配置，将其保存为expand-cluster-reassignment.json
2. 执行扩容命令： bin/kafka-reassign-partitions.sh --zookeeper 10.16.10.76:2181/kafka-0.8.1 --reassignment-json-file expand-cluster-reassignment.json --execute  正常执行的话会生成同上图类似的json串，表示原始状态和目标状态
3. 查询执行状态： bin/kafka-reassign-partitions.sh --zookeeper 10.16.10.76:2181/kafka-0.8.1 --reassignment-json-file expand-cluster-reassignment.json --verify  正常执行后会返回当前数据迁移的不用partion的，信息状态类似下面
   

   Reassignment of partition [foo1,0] completed successfully   //移动成功
   Reassignment of partition [foo1,1] is in progress          //这行代表数据在移动中
   Reassignment of partition [foo1,2] is in progress
   Reassignment of partition [foo2,0] completed successfully
   Reassignment of partition [foo2,1] completed successfully 
   Reassignment of partition [foo2,2] completed successfully 

4. 数据迁移一旦开始无法停止，也不要强行停止集群，这样会造成数据不一致，带来无法挽回的后果。

5. 注意：kafka数据迁移的原理是先拷贝数据到目标节点，然后再删除原节点的数据。这样的话如果集群原节点空间不足，不要继续指定其为迁移broker，这样将造成原节点空间用尽，例如原节点是broker为0，1，2，3，4就不要这样指定 --broker-list "0,1,2,3,4,5,6,7,8"，应该这样 --broker-list "5,6,7,8"。
   另外数据迁移也可以通过手工定制。
6. 节点leader rebalance(移动备份数据后，重新平衡领导者分配)：

   bin/kafka-preferred-replica-election.sh --zookeeper zk_host:port/chroot

7.运维问题总结

1. 操作系统版本最好保持一致