Kafka部署过程记录

0-前言

Kafka是一种高吞吐量的分布式消息发布订阅系统，由Scala语言编写，其分布式特性基于Zookeeper实现。需要指出的是Kafka虽然也是类似于消息队列发布订阅的使用模式，但其并不完全遵循JMS协议。

1-集群原理

结构.png

上图展示了Kafka中各个元素及互相之间的关联关系，下面就几个重点部分进行说明：

• Topic和Partition
  Topic用于区分不同的消息组，每一条消息都必须对应一个Topic。在Kafka中一个Topic又可以被分为若干个Partition，每个Partition还可以有一个或多个副本构成leader/follower关系，这些Partition分布在不同的Broker节点之上。需要注意的是，Producer 和 Consumer 只与 leader Partition进行交互，该leader宕机后会由Controller重新进行选举。
  Partition的状态有四种：

1. NonExistentPartition -- 表示该分区要么没有被创建过或曾经被创建过但后面被删除了
2. NewPartition -- 分区创建之后就处于NewPartition状态。在这个状态中，分区已经分配了副本，但是还没有选举出leader和ISR
3. OnlinePartition -- 一旦分区的leader被推选出来，它就处于OnlinePartition状态
4. OfflinePartition -- 如果leader选举出来后，leader宕机了，那么该分区就处于OfflinePartition状态

• Consumer
  每一个Consumer都属于一个Consumer Group，同一个Group下都会拥有相同的GroupId。Kafka中消息的消费是以Consumer Group为单位的，简单来说就是一条消息会被每一个Group各消费一次，而同一个Group中只会有一个Consumer消费该消息。由于一个Partition只能被每个Group中的一个Consumer消费（一个Consumer可以消费多个Partition），因此每个Group中Consumer的数量不应多于Partition的数量。
• Controller
  Kafka集群中的其中一个Broker会被选举为Controller，主要负责Partition管理和副本状态管理，也会执行类似于重分配Partition之类的管理任务。如果当前的Controller宕机，会从其他正常的Broker中重新选举Controller。
• ISR
  leader Partition会在Zookeeper中维护一个与其基本保持同步的Replica列表，该列表称为ISR(in-sync Replica)，每个Partition都会有一个ISR，由leader动态维护。可以简单认为ISR是一个记录follower的列表，如果一个follower比leader落后太多，或者超过一定时间未发起数据复制请求，则leader会将其从ISR中移除。当ISR中所有Replica都向leader发送ACK时，leader才会进行commit操作。

2-集群搭建

Kafka部署需要准备以下环境：

1. java环境
2. Gradle环境
3. Zookeeper
   操作系统为Centos6.9，java采用jdk1.8，Zookeeper版本3.6.4，由于这些都是已安装的所以在此不做安装说明，只说一下Gradle的安装

2.1-Gradle安装

我安装的是4.10.2版本，首先对安装包进行解压

unzip -d /opt/gradle gradle-4.10.2-bin.zip

打开/etc/profile配置环境变量path

export PATH=$PATH:/opt/gradle/gradle-4.10.2/bin

通过以下命令检查是否配置成功

gradle -v

看到如下信息说明已成功

------------------------------------------------------------
Gradle 4.10.2
------------------------------------------------------------

Build time:   2018-09-19 18:10:15 UTC
Revision:     b4d8d5d170bb4ba516e88d7fe5647e2323d791dd

Kotlin DSL:   1.0-rc-6
Kotlin:       1.2.61
Groovy:       2.4.15
Ant:          Apache Ant(TM) version 1.9.11 compiled on March 23 2018
JVM:          1.8.0_181 (Oracle Corporation 25.181-b13)
OS:           Linux 2.6.32-696.13.2.el6.x86_64 amd64

2.2-Kafka安装

Kafka集群使用了三台服务器，每台上单独部署一个Kafka节点，Kafka版本2.0.0
首先进行解压

tar -xzf kafka_2.11-2.0.0.tgz

进入Kafka目录

cd kafka_2.11-2.0.0

修改配置文件

vi config/server.properties

其中有几个重要参数如下

broker.id=0
advertised.listeners=PLAINTEXT://ip:9092
zookeeper.connect=ip1:2181,ip2:2181,ip3:2181
log.dirs=/tmp/kafka-logs
log.retention.hours=168
num.partitions=3

broker.id可以随便设置，但必须保证同一集群下的节点不重复；zookeeper.connect是zookeeper的地址，需要按实际填写；advertised.listeners是对外展示的地址和端口（端口默认使用9092），在后面配置管理页面的时候有用；log.dirs和log.retention.hours表示日志的存放路径和保存时间；num.partitions是Topic下默认分配的Partition数量。
接着修改一下Kafka的启动配置

vi bin/kafka-server-start.sh

文件中增加下述内容

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export JMX_PORT="9999"
fi

这里设置了堆的大小以及JMX监控端口，方便后续使用监控页面。

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

以上设置都完成后启动Kafka

2.3-监控页面安装

Kafka官方没有提供可视化的监控页面，但有不少第三方的开源产品可供选择，比较主流的有以下三款

• KafkaOffsetMonitor
• Kafka web console
• Kafka Manager

最终我选择的是yahoo开源的Kafka Manager，项目地址https://github.com/yahoo/kafka-manager。
首先下载源码包kafka-manager-1.3.3.18.tar.gz，这里安装的是最新的1.3.3.18版

tar -zxf kafka-manager-1.3.3.18.tar.gz
cd kafka-manager-1.3.3.18
./sbt clean dist

依次执行上述三条命令，解压并进入目录，生成安装包。Kafka-manager使用了Play框架，如果之前从来没有用过Play框架，下载依赖的jar包会花费较长的时间(真的不是一般的久，我花了2个小时。。。)，想要加快这个过程可以将源配置为aliyun的私服。
操作如下，通过cd ~进入当前用户目录，然后通过命令mkdir .sbt创建.sbt目录，进入创建的该目录，使用vi创建repositories文件，编辑内容如下：

[repositories]
local
aliyun: http://maven.aliyun.com/nexus/content/groups/public
typesafe: http://repo.typesafe.com/typesafe/ivy-releases/, [organization]/[module]/(scala_[scalaVersion]/)(sbt_[sbtVersion]/)[revision]/[type]s/[artifact](-[classifier]).[ext], bootOnly

待命令执行完成后，在target/universal 目录中会生成一个zip压缩包kafka-manager-1.3.3.18.zip。
解压并进入该压缩包

unzip kafka-manager-1.3.3.18.zip
cd kafka-manager-1.3.3.18

修改配置信息

vi conf/application.conf

修改配置文件中的zookeeper相关配置项kafka-manager.zkhosts，修改完成后通过下述命令启动工程

bin/kafka-manager 

启动默认使用9000端口，若要指定其它端口也可以通过以下命令启动

bin/kafka-manager -Dconfig.file=/path/to/application.conf -Dhttp.port=9000

接着便可以通过http://ip:9000地址访问

Kafka manager.png

3-总结

本文介绍了Kafka的集群原理，记录了Kafka集群的部署过程，以及可视化监控页面kafka manager的安装过程。