第2章 Kafka安装配置

第2章 Kafka安装配置

本章将详细介绍Kafka运行环境的搭建,包括在Linux系统和Windows系统中搭建Kafka运行环境。

2.1 基础环境配置

由于Kafka是用Scala语言开发的,运行在JVM上,因此在安装Kafka之前需要先安装JDK。

2.1.1 JDK安装配置

最新版本的Kafka需要运行在JDK 1.7以上,Kafka官方网站推荐使用JDK 1.8,因此本书所应用的JDK环境采用JDK 1.8。下面将详细介绍JDK 1.8安装步骤。

1.Windows安装JDK

(1)下载并安装。首先在Oracle官方网站http://www.oracle.com/technetwork/java/javase/ downloads/jdk8-downloads-2133151.html下载JDK 1.8安装文件,根据操作系统类型选择相应的JDK版本。我使用的是64位操作系统,因此下载jdk-8u111-windows-x64.exe安装文件。下载完成后,双击运行安装。在安装时可以选择安装路径,这里在安装时全使用默认路径。

(2)环境变量配置。在系统变量中新增变量名JAVA_HOME,变量值为JDK 1.8安装路径。由于Java默认安装在Program Files目录下,这个目录名之间有空格,有可能在运行某些应用时因JDK安装路径有空格而报错。例如,我在安装JDK后,运行Kafka时报如下错误:

错误: 找不到或无法加载主类 Files\Java\jdk1.8.0_111\lib\dt.jar;C:\Program

为了避免出现类似的错误,在Windows系统上,若JDK安装在Program File目录下,在设置JAVA_HOME时,用该目录别名PROGRA~1,因此将JAVA_HOME设置为C:\PROGRA~1\Java\jdk1.8.0_111。新增变量名CLASSPATH,变量值为.;% JAVA_HOME%\lib\dt.jar;% JAVA_HOME%\ lib\tools.jar。本步操作JDK环境变量配置如表2-1所示。

表2-1 JDK环境变量配置

变 量 类 型

变 量 名

变 量 值

系统变量

JAVA_HOME

C:\PROGRA~1\Java\jdk1.8.0_111

用户变量

CLASSPATH

.;% JAVA_HOME%\lib\dt.jar;%JAVA_HOME%\lib\tools.jar

(3)验证。环境变量配置完成后,在Windows的cmd终端输入查看Java版本的命令,以此来验证JDK安装配置是否成功。命令如下:

Java –version

若输出为以下JDK版本信息,则表示JDK 1.8已安装成功,且为系统默认JDK。

Java version "1.8.0_111"
Java(TM) SE Runtime Environment (build 1.8.0_111-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.111-b14, mixed mode)

2.Linux安装JDK

一些Linux的发行版默认已安装了JDK,如OpenJDK,这里所用的Linux操作系统默认已经安装了OpenJDK。输入下面的命令查看JDK版本信息:

Java  –version

输出JDK版本信息如下:

Java version "1.7.0_45"
OpenJDK Runtime Environment (rhel-2.4.3.3.el6-x86_64 u45-b15)
OpenJDK 64-Bit Server VM (build 24.45-b08, mixed mode)

然而有些Linux系统并没有安装JDK,因此本小节将详细讲解如何在Linux中安装JDK。这里我们讲解JDK 1.8的安装。

(1)下载并安装。进入Oracle官方网站http://www.oracle.com/technetwork/Java/Javase/downloads/ jdk8-downloads-2133151.html下载Linux版本的JDK 1.8安装包。这里我们下载的安装包版本为jdk-8u111-linux-x64.gz,并将安装包解压到/usr/local/software/Java路径下。

tar -xzvf jdk-8u111-linux-x64.gz      # 解压jdk安装包

将安装包解压后,即完成JDK的安装。

(2)配置环境变量。在/etc/profile文件中添加JDK和JRE的路径,并添加到Path中,操作命令如下:

vi /etc/profile      # 编辑profile文件

在文件中添加以下内容:

export JAVA_HOME=/usr/local/software/Java/jdk1.8.0_111
export JRE_HOME=/usr/local/software/Java/jdk1.8.0_111/jre
export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin

保存退出。若系统默认安装了OpenJDK,则用户可以选择将其删除,也可以修改配置用最新安装的版本替换OpenJDK。这里选择用新安装的JDK替换系统自带的OpenJDK,则按序执行以下命令。

  • 将Java添加到bin:
    update-alternatives --install /usr/bin/Java Java /usr/local/software/Java/jdk1.8.0_111 300
  • 将Javac添加到bin:
    update-alternatives --install /usr/bin/Javac Javac /usr/local/software/   
         Java/jdk1.8.0_111/bin/Javac 300
  • 选择JDK版本:
    update-alternatives --config Java

执行以上命令会出现JDK版本选择界面,如图2-1所示。这里我们选择新安装的JDK 1.8,即输入序号3,按回车键。

图2-1 Linux控制台展示的JDK版本选择命令行界面

环境变量配置好后执行以下命令,让刚才的修改操作立即生效:

    source /etc/profile           # 让对/etc/profile的修改立即生效

(3)验证。输入查看JDK版本命令,查看环境变量配置是否成功,执行以下命令:

Java –version      # 查看jdk版本

输出以下JDK版本信息:

Java version "1.8.0_111"
Java(TM) SE Runtime Environment (build 1.8.0_111-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.111-b14, mixed mode)

由以上JDK版本信息可知,JDK版本已替换为新安装的JDK 1.8版本。至此,JDK安装完成。

2.1.2 SSH安装配置

对Kafka集群本身来讲,配置SSH免密钥登录并不是必需的步骤,但作为分布式系统,一般会由多台机器构成。为了便于操作管理,如通过SSH方式启动集群代理等,这里对SSH安装配置进行介绍。

(1)在根目录下查看是否存在一个隐藏文件夹.ssh。若没有该文件夹,则在确保机器联网条件下执行以下命令安装ssh:

sudo apt-get install ssh      # 安装ssh

(2)进入.ssh目录,生成密钥对,执行命令如下:

ssh-keygen -t rsa      # 产生密钥

在执行以上命令时一路回车即可。ssh-keygen用于生成认证密钥,-t用来指定密钥类型,这里选择rsa密钥。执行完毕后会在~/.ssh目录下生成id_rsa和id_rsa.pub两个文件,其中id_rsa为私钥文件,id_rsa.pub为公钥文件。依次在集群其他机器上完成步骤1和步骤2。

(3)将id_rsa.pub文件内容追加到授权的key文件中,命令如下:

cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys      # 追加公钥到授权文件中

若是单机环境,则至此已完成ssh配置。

(4)将第一台机器的authorized_keys文件复制到第二台机器上,并将第二台机器的公钥也追加到authorized_keys文件中,依次执行以下命令:

scp authorized_keys root@172.117.12.62:~/.ssh/   # 复制第一台机器的授权文件到第二台机器
cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys  # 在第二台机器上执行此命令,将第二台机器的公钥追加到授权文件中

(5)将第二台机器的authorized_keys文件复制到第三台机器上,并将第三台机器的公钥追加到authorized_keys文件中,执行命令如下:

scp authorized_keys root@172.117.12.63:~/.ssh/   # 复制第二台机器的授权文件到第三台机器
cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys   # 在第三台机器上执行此命令,将第三台机器的公钥追加到授权文件中

若集群还有更多机器,则依此类推完成授权文件合并。至此 ssh 配置完成,在已配置 ssh的任何一台机器上均可免密登录到其他机器。例如,在第一台机器上执行以下ssh命令,输出如下(首次登录会让输入密码):

[root@rhel65 .ssh]# ssh 172.117.12.61
The authenticity of host '172.117.12.61 (172.117.12.61)' can't be established.
RSA key fingerprint is a3:5b:a9:29:ed:00:74:f4:ce:51:e5:7c:42:5b:8d:44.
Are you sure you want to continue connecting (yes/no)? yes
Warning: Permanently added '172.117.12.61' (RSA) to the list of known hosts.
root@172.117.12.61's password: 
Last login: Wed Feb  8 17:30:11 2017 from server-1
[root@rhel65 ~]# ssh 172.117.12.61
Last login: Wed Feb  8 17:32:04 2017 from server-3
[root@rhel65 ~]# ssh 172.117.12.62
Last login: Wed Feb  8 17:26:09 2017 from server-1

2.1.3 ZooKeeper环境

ZooKeeper是一个分布式应用程序协调服务框架,分布式应用程序可以基于ZooKeeper来实现同步服务、配置维护、命名服务等,ZooKeeper能提供基于类似于文件系统的目录节点树方式的数据存储,通过监控各节点数据状态的变化,达到基于数据的集群管理。ZooKeeper主要由表2-2所示的几个角色构成。

表2-2 ZooKeeper集群主要角色说明

角  色

 

描  述

Leader

 

集群的领导者,负责投票的发起和决议及更新系统状态

Learner

Follower

跟随者,接受客户端的请求并返回结果给客户端,参与投票

 

Observer

接受客户端的请求,将写的请求转发给Leader,不参与投票。Observer目的是扩展系统,提高读的速度

关于ZooKeeper的原理及其他相关知识,读者可查阅ZooKeeper官方网站(http://mirrors. cnnic.cn/apache/zookeeper/)及相关书籍进行深入了解。

Kafka依赖ZooKeeper,通过ZooKeeper来对代理、消费者上下线管理、集群、分区元数据管理等,因此ZooKeeper也是Kafka得以运行的基础环境之一。

进入ZooKeeper官方网站http://mirrors.cnnic.cn/apache/zookeeper/下载ZooKeeper(本书所用ZooKeeper版本为zookeeper-3.4.8),然后将下载文件解压到指定目录。对ZooKeeper的安装,下面按Windows和Linux分别进行讲解。

1.Windows安装ZooKeeper

一般会选择在Linux操作系统上安装和部署分布式服务,因此这里并不打算讲解Windows环境下ZooKeeper集群环境搭建,只是简单介绍Windows环境下ZooKeeper单机模式的安装。

(1)解压安装。首先将ZooKeeper安装包zookeeper-3.4.8.tar.gz解压到相应目录,这里将ZooKeeper解压到D:\software\zookeeper-3.4.8目录下。然后进入ZooKeeper安装路径conf目录下,会看到ZooKeeper提供了一个zoo_sample.cfg的配置模板,将该文件重命名为zoo.cfg。zoo.cfg文件中只需修改dataDir和dataLogDir配置,其他配置使用默认值(其他配置及其含义将在下面的“Linux搭建ZooKeeper环境”小节详细介绍)。这里对dataDir和dataLogDir配置如下:

dataDir=F:\\zookeeper\\data
dataLogDir=F:\\zookeeper\\logs

至此,Windows环境下ZooKeeper安装配置完成。下面进入ZooKeeper安装路径bin目录下,启动及验证ZooKeeper安装是否成功。

(2)验证。执行启动ZooKeeper命令:

zkServer.cmd    # windows下启动ZooKeeper

若输出没有任何错误,通过jps命令可以看到ZooKeeper相关进程。输入命令:

jps  # 查看Java进程命令

输出结果中至少包括以下进程名:

12008 QuorumPeerMain
11596 Jps

还可以进入ZooKeeper的安装路径bin目录下,通过ZooKeeper客户端连接到ZooKeeper服务,执行以下命令进一步验证ZooKeeper是否安装成功:

zkCli.cmd -server 127.0.0.1:2181   # 登录到ZooKeeper服务器

在输出信息中会看到“Welcome to ZooKeeper!”,同时显示接受命令输入界面。

在客户端输入:

ls /      # 查看ZooKeeper服务器目录结构

此时ZooKeeper服务器中仅有一个zookeeper节点,信息显示如下:

[zk: 127.0.0.1:2181(CONNECTED) 0] ls /
[zookeeper]

至此,Windows环境下安装ZooKeeper讲解完毕。

2.Linux搭建ZooKeeper环境

在Linux环境下ZooKeeper单机模式配置与上一小节介绍的Windows环境下ZooKeeper安装配置的操作步骤基本相同,因此本小节直接介绍ZooKeeper分布式环境搭建。下面将讲解在Linux环境下如何配置由3台机器构成的ZooKeeper集群环境,这3台机器的IP地址分别为172.117.12.61、172.117.12.62和172.117.12.63。

(1)解压安装。首先在3台机器上分别将zookeeper-3.4.8.tar.gz解压到/usr/local/software/ zookeeper目录。进入解压后的zookeeper-3.4.8 /conf目录,将zoo.sample.cfg重命名为zoo.cfg。关于ZooKeeper配置文件中几个基础配置项的说明如表2-3所示。

表2-3 ZooKeeper基础配置说明

配 置 项

默 认 值

说  明

tickTime

2000ms

ZooKeeper中的一个时间单元。ZooKeeper中所有时间都以这个时间单元为基准,进行整数倍配置,默认是2 s

initLimit

10

Follower在启动过程中,会从Leader同步所有最新数据,确定自己能够对外服务的起始状态。当Follower在initLimt个tickTime还没有完成数据同步时,则Leader仍为Follower连接失败

syncLimit

5

Leader与Follower之间通信请求和应答的时间长度。若Leader在syncLimit个tickTime还没有收到Follower应答,则认为该Leader已下线

dataDir

/tmp/zookeeper

存储快照文件的目录,默认情况下,事务日志也会存储在该目录上。由于事务日志的写性能直接影响ZooKeeper性能,因此建议同时配置参数dataLogDir

dataLogDir

/tmp/zookeeper

事务日志输出目录

clientPort

2181

ZooKeeper对外端口

请读者根据自已服务器环境,修改zoo.cfg文件中表2-3提及参数的配置。这里只修改了以下两个配置项,其他几个基础配置沿用默认值。

dataDir=/opt/data/zookeeper/data
dataLogDir=/opt/data/zookeeper/logs

若是单机模式,操作至此完成。接下来配置将3台机器构成一个分布式集群。

(2)集群配置。首先在3台机器的/etc/hosts文件中加入3台机器的IP与机器域名映射,域名自定义,这里分别命名为server-1、server-2、server-3,3台机器IP与机器域名映射关系如下:

172.117.12.61 server-1
172.117.12.62 server-2
172.117.12.63 server-3

然后进入其中一台机器的ZooKeeper安装路径conf目录。这里我们选择在IP为172.117.12.61的机器上进行配置,编辑conf/zoo.cfg文件,在该文件中添加以下配置:

server.1=server-1:2888:3888
server.2=server-2:2888:3888
server.3=server-3:2888:3888

为了便于讲解以上配置,在这里抽象一个公式,即server.n=n-server-domain:port1:port2。这个公式中的n是一个数字类型常量,这里配置的1、2和3用于表示第几台ZooKeeper服务器;n-server-domain表示第n台ZooKeeper服务器的IP所映射的域名,当然这里也可以是第n台机器的IP;port1表示该服务器与集群中的Leader交换信息的端口,默认是2888;port2表示选举时服务器相互通信的端口。

接着在${dataDir}路径下创建一个myid文件。myid里存放的值就是服务器的编号,即对应上述公式中的n,在这里第一台机器myid存放的值为1。ZooKeeper在启动时会读取myid文件中的值与zoo.cfg文件中的配置信息进行比较,以确定是哪台服务器。

在zoo.cfg文件中我们同时配置了3台机器,因此接下来通过scp命令将本台机器的zoo.cfg文件复制到另外两台机器相应目录进行替换。

scp zoo.cfg root@172.117.12.62:/usr/local/software/zookeeper/zookeeper-3.4.8/conf/
scp zoo.cfg root@172.117.12.63:/usr/local/software/zookeeper/zookeeper-3.4.8/conf/

然后分别修改另外两台机器的myid。同时,为了操作方便,我们将ZooKeeper相关环境变量添加到/etc/profile文件当中。

设置ZooKeeper安装路径,在/etc/profile相关环境变量配置中添加以下信息:

export ZOOKEEPER_HOME=/usr/local/software/zookeeper/zookeeper-3.4.8

在该文件的Path配置项最后加上:$ZOOKEEPER_HOME/bin。注意,在$ZOOKEEPER_HOME前有一个冒号。然后执行source/etc/profile命令使所做的修改操作立即生效。其他两台机器也进行同样的环境设置。至此,由3台机器构成的分布式ZooKeeper环境搭建步骤介绍完毕。下面启动ZooKeeper进行验证。

(3)验证。由于配置了ZooKeeper环境变量,因此无需进入ZooKeeper安装路径bin目录下。在这3台机器上分别启动ZooKeeper:

zkServer.sh start    # 启动ZooKeeper服务

输出如下信息:

ZooKeeper JMX enabled by default
Using config: /usr/local/software/zookeeper/zookeeper-3.4.8/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

查看这3台ZooKeeper服务器状态,依次在这3台机器上执行以下命令:

zkServer.sh status    # 查询zookeeper状态

执行上述启动命令,其中有两台机器输出以下信息:

ZooKeeper JMX enabled by default
Using config: /usr/local/software/zookeeper/zookeeper-3.4.8/bin/../conf/zoo.cfg
Mode: follower

另外一台机器输出信息如下:

ZooKeeper JMX enabled by default
Using config: /usr/local/software/zookeeper/zookeeper-3.4.8/bin/../conf/zoo.cfg
Mode: leader

可以看到,这3台机器中,一台机器作为Leader,其他两台服务器作为Follower。同时,可以查看zookeeper.out文件内容,通过启动日志进一步了解ZooKeeper运行过程。至此,ZooKeeper集群环境搭建讲解完毕。

2.2 Kafka单机环境部署

Kafka安装较简单,不同操作系统下安装步骤基本相同,针对大多数用户来讲,在生产环境使用Kafka一般选择Linux服务器,本书Kafka实战操作也是基于Linux环境进行讲解的。下面分别介绍Kafka在Windows操作系统以及Linux操作系统的安装步骤,Mac操作系统的安装步骤与Linux操作系统的安装步骤基本类似,不再介绍。同时,后面几节中的Kafka集群环境搭建也只介绍在Linux环境下Kafka集群环境的部署。

2.2.1 Windows环境安装Kafka

Windows下安装Kafka只需将下载的Kafka安装包解压到相应目录即可。

(1)下载及安装。进入Kafka官方网站http://kafka.apache.org/downloads下载当前最新版本的Kafka,Kafka安装包并没有区分Windows安装包还是Linux安装包,仅在bin目录下将Windows环境执行Kafka的相关脚本放在/bin/windows目录下。当前Kafka最新版本为kafka_2.11-0.10.1.1.tgz,其中2.11代表Scala版本,0.10.1.1表示Kafka的版本。这里将下载的安装包解压到D:\software\kafka_2.11-0.10.1.1目录下,为了便于讲解,这里记Kafka安装路径为$KAFKA_HOME。至此,Windows下Kafka完成安装。当然我们也可以像安装JDK一样配置Kafka环境变量,感兴趣的读者可以自行配置,这步操作不是必需的步骤,因此不再阐述。

(2)启动KafkaServer验证。安装好Kafka后,启动KafkaServer。在启动Kafka之前,需要启动Zoookeeper。若ZooKeeper服务不是本地服务,应修改Kafka安装目录下/config/server.properties文件zookeeper.connect配置项,然后在Windows的cmd下进入$KAFKA_HOME/bin/windows目录,执行以下命令,启动KafkaServer。

kafka-server-start.bat ../../config/server.properties   # windows下启动kafak server

若在启动过程中没有报任何异常信息,同时在控制台最后输出打印内容如图2-2所示,则表示Kafka在Windows环境下安装成功。

图2-2 KafkaServer启动日志

2.2.2 Linux环境安装Kafka

在Linux系统上安装Kafka与在Windows系统上安装操作基本相同,将安装包解压到相应目录,这里依然将Kafka安装目录记为$KAFKA_HOME,修改$KAFKA_HOME/config/server.properties文件相关配置即可。这里安装Kafka所用机器与安装ZooKeeper的机器相同,但在生产环境,一般将ZooKeeper集群与Kafka机器分机架部署。在讲解Kafka单机版本安装时,我们选择3台机器中的一台,IP为172.117.12.61。

(1)解压安装。先将Kafka安装包kafka_2.11-0.10.1.1.tgz解压到指定目录下,这里将Kafka解压到/usr/local/software/kafka目录下。进入/usr/local/software/kafka目录执行以下命令解压Kafka安装包。

tar -xzvf kafka_2.11-0.10.1.1.tgz      # 解压安装Kafka

由于后续对Kafka的讲解都是在Linux环境下,因此为了操作方便我们对Kafka的环境变量进行设置。在/etc/profile文件中加入Kafka安装路径,并将Kafka的bin目录添加进Path中。这一步操作并非Kafka安装必需的设置,读者可根据情况选择是否需要对Kafka环境变量进行配置。打开/etc/profile文件添加以下配置。

  • 指定Kafka安装路径:
    export KAFKA_HOME=/usr/local/software/kafka/kafka_2.11-0.10.1.1
  • 将Kafka bin目录加到Path。在Path设置后添加:$KAFKA_HOME/bin,添加Kafka bin后完整的Path如下:
    export PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin:$KAFKA_HOME/bin

保存文件退出,执行source /etc/profile命令让刚才新增的Kafka环境变量设置生效。再在任一路径下输入kafka然后按Tab键,会提示补全Kafka运行相关脚本.sh文件,表示Kafka环境变量配置成功,但一般Kafka脚本运行时会加载/config路径下的相关配置文件,因此当不在Kafka安装目录bin下执行相关脚本时,需要指定配置文件绝对路径。

(2)修改配置。修改$KAFKA_HOME/config目录下的server.properties文件,为了便于后续集群环境搭建的配置,需要保证同一个集群下broker.id要唯一,因此这里手动配置broker.id,直接保持与ZooKeeper的myid值一致,同时配置日志存储路径。server.properties修改的配置如下:

broker.id=1                      # 指定代理的id
log.dirs=/opt/data/kafka-logs    # 指定Log存储路径

其他配置保持不变,由于172.117.12.61这台机器本地已安装了ZooKeeper,因此在Kafka单机版本安装讲解时,我们暂不对zookeeper.connect配置进行修改,其他配置文件也暂不进行修改。

(3)验证

  • 启动Kafka,要保证ZooKeeper已正常启动,进入Kafka安装路径$KAFKA_HOME/ bin目录下,执行启动KafkaServer命令。
    kafka-server-start.sh -daemon ../config/server.properties    # 启动Kafka

执行jps命令查看Java进程,此时进程信息至少包括以下几项:

    15976 Jps
    14999 QuorumPeerMain
    15906 Kafka

可以看到ZooKeeper进程和Kafka进程名,同时进入$KAFKA_HOME/logs目录下,查看server.log会看到KafkaServer启动日志,在启动日志中会记录KafkaServer启动时加载的配置信息。

  • 通过ZooKeeper客户端登录ZooKeeper查看目录结构,执行以下命令:
    zkCli.sh -server server-1:2181    # 登录ZooKeeper 
    ls /                              # 查看ZooKeeper目录结构

在Kafka启动之前ZooKeeper中只有一个zookeeper目录节点,Kafka启动后目录节点如下:

    [cluster, controller, controller_epoch, brokers, zookeeper, admin, isr_change_   
        notification, consumers, config]

执行以下命令,查看当前已启动的Kafka代理节点:

    ls /brokers/ids      # 查看已启动的代理节点
     [1]                  # 已启动的代理节点对应的brokerId

输出信息显示当前只有一个Kafka代理节点,当前代理的brokerId为1。至此,Kafka单机版安装配置介绍完毕,相关操作我们将在第5章进行详细介绍。

2.3 Kafka伪分布式环境部署

KafkaServer启动时需要加载一个用于KafkaServer初始化相关配置的server.properties文件,当然文件名可以任意,一个server.properties对应一个KafkaServer实例。Kafka伪分布式就是在一台机器上启动多个KafkaServer来达到多代理的效果,因此要保证broker.id及port在同一台机器的多个server.properities中唯一。

本节在上一节的Linux安装Kafka基础配置之上,将server.properties文件复制一份并命名为server-0.properties,在server-0.properties文件中修改配置如下:

broker.id=0
log.dirs=/opt/data/kafka-logs/broker-0
port=9093

由于代理默认端口是9092,server.properties没有设置端口则采用默认设置,因此在server-0.properties将port设置为9093。这个端口可以自定义,只要新端口没有被占用即可。执行以下命令,分别启动brokerId为0和1的两个KafkaServer:

kafka-server-start.sh -daemon ../config/server-0.properties      # 启动代理0
kafka-server-start.sh -daemon ../config/server.properties        # 启动代理1

再次执行jps命令查看Java进程信息,打印输出如下信息:

19453 Kafka
18036 ZooKeeperMain
18228 QuorumPeerMain
19169 Kafka
19504 Jps

从输出的进程信息可以看到有两个Kafka进程存在,即代表刚才启动的broker.id为0和1的两个代理。此时登录ZooKeeper客户端,再查看ZooKeeper的/brokers/ids目录,会看到该目录下有两个节点:

[zk: 172.117.12.61(CONNECTED) 0] ls /brokers/ids
[0, 1]

这样,一台机器上启动多个代理的伪分布式环境安装配置介绍完毕。

2.4 Kafka集群环境部署

2.2.2节已经讲解了Kafka单机版安装配置,因此对Kafka集群环境配置时只需将单机版安装的Kafka配置进行相应修改,然后复制到另外两台机器即可。这里只需修改server.properties文件中Kafka连接ZooKeeper的配置,将Kafka连接到ZooKeeper集群,配置格式为“ZooKeeper服务器IP:ZooKeeper的客户端端口”,多个ZooKeeper机器之间以逗号分隔开。

zookeeper.connect=server-1:2181,server-2:2181,server-3:2181

进入172.117.12.61服务器/usr/local/software目录下,执行以下两条命令将本机安装的Kafka分别复制到另外两台服务器上:

scp -r kafka_2.11-0.10.1.1  root@172.117.12.62:/usr/local/software/kafka
scp -r kafka_2.11-0.10.1.1  root@172.117.12.63:/usr/local/software/kafka

复制完成后,分别登录另外两台机器,修改server.properties文件中的broker.id依次为2和3。当然可以设置任一整数,只要保证一个集群中broker.id唯一即可。同时在3台机器的server.properties文件中设置host.name为本机的IP。例如,对主机名为server-1的机器上的Kafka节点,在server.properties文件中增加host.name=172.117.12.61。本书所用版本的Kafka若不设置host.name,则会在创建主题及向主题发送消息时发生NOT_LEADER_FOR_PARTITION这样的异常。

配置完毕后,分别启动3台机器的KafkaServer,通过ZooKeeper客户端查看Kafka在ZooKeeper中的相应元数据信息,其中查看/brokers/ids节点信息如下:

[zk: 172.117.12.61(CONNECTED) 1] ls /brokers/ids
[1, 2, 3]

由/brokers/ids节点存储的元数据可知,3台机器的Kafka均已正常已启动。至此,Kafka分布式环境搭建过程介绍完毕。

2.5 Kafka Manager安装

在实际应用中,我们经常需要了解集群的运行情况,如查看集群中代理列表、主题列表、消费组列表、每个主题对应的分区列表等,抑或是希望通过简单的Web界面操作来创建一个主题或是在代理负载不均衡时,手动执行分区平衡操作等。为了方便对Kafka集群的监控及管理,目前已有开源的Kafka监控及管理工具,如Kafka Manager、Kafka Web Console、KafkaOffsetMonitor等,读者也可以根据自己业务需要进行定制开发。本节只简单讲解Kafka Manager的安装应用。

Kafka Manager由yahoo公司开发,该工具可以方便查看集群主题分布情况,同时支持对多个集群的管理、分区平衡以及创建主题等操作。读者可访问https://github.com/yahoo/kafka- manager进行深入了解。

(1)下载编译Kafka Manager。进入GitHut官网搜索关键词“kafka-manager”即可查询到Kafka Manager的下载地址,具体地址为https://github.com/yahoo/kafka-manager/,直接点击“Clone or download”按钮进行下载。将下载的kafka-manager-master.zip文件上传到Linux服务器。用户也可以在Linux机器上执行以下命令在线下载Kafka Manager源码:

git clone https://github.com/yahoo/kafka-manager  # 从GitHub上下载Kafka Manager源码

Kafka Manager是用Scala语言开发的,通过sbt(Simple Build Tool)构建,sbt是对Scala或Java语言进行编译的一个工具,它类似于Maven,Gradle。截止到编写本书时,Kafka Manager是基于0.9.0.1版本的Kafka开发的,鉴于Kafka 0.9与Kafka-0.10版本的实现,该版本的Kafka Manager也能作为0.10.+版本的Kafka管理及监控工具,在Kafka Manager管理界面添加集群管理配置时,Kafka Version选0.9.0.1即可。待源码下载之后,进入Kafka Manager源码目录,会有一个sbt文件,执行以下命令进行Kafka Manager源码编译。

./sbt clean dist      # 编译Kafka Manager源码

编译过程会下载相关的jar文件,因此有些耗时。等源码编译完成后,在控制台输出的编译日志的最后几行信息如下:

[info] Your package is ready in /home/morton/.sbt/0.13/staging/17dfe5a6b216985c290a/   
kafka-manager-master/target/universal/kafka-manager-1.3.2.1.zip
[info] [success] Total time: 170 s, completed 2017-1-15 14:23:45

从控制台输出的编译日志信息可以看到,在编译时会在/home/用户名/路径下创建一个.sbt目录,编译后的文件存放在该目录相应子目录里,编译日志信息中的 morton 为编译 Kafka Manager源码的机器名。在编译过程中出现:

Download failed. Obtain the jar manually and place it at /home/morton/.sbt/launchers/   
0.13.9/sbt-launch.jar

表示在编译过程下载sbt-launch.jar文件遇到问题,请读者单独下载sbt-launch.jar相应版本并上传到/home/用户名/.sbt/launchers/0.13.9/目录下,再次执行编译命令。最终会在/home/用户名/.sbt/0.13/staging相应子目录下生成kafka-manager-1.3.2.1.zip文件,该文件就是用来对Kafka进行监控和管理的工具。若读者在编译时由于个人网络环境原因无法编译,可以直接在网络上下载该文件然后复制到服务器。将编译好的kafka-manager-1.3.2.1.zip文件解压到指定位置(这里解压到/usr/local/software/kafka-manager目录下)即完成安装。

(2)修改配置。进入Kafka Manager安装路径下的conf目录,打开application.conf文件,修改以下配置。将kafka-manager.zkhosts="kafka-manager-zookeeper:2181"配置项,修改为实际的ZooKeeper连接地址,例如这里修改为:

kafka-manager.zkhosts="172.117.12.61:2181,172.117.12.62:2181,172.117.12.63:2181"

(3)启动Kafka Manager。进入bin目录下执行以下启动命令:

nohup ./kafka-manager -Dconfig.file=../conf/application.conf &   # 启动Kafka Manager

Kafka Manager默认请求端口是9000,在浏览器中输入安装Kafka Manager服务地址及9000端口访问Kafka Manager,如访问http://172.117.12.62:9000。Kafka Manager启动初始化界面如图2-3所示。

图2-3 Kafka Manager启动初始化界面

通过修改配置文件application.conf里http.port的值,或是通过命令行参数传递可以修改Kafka Manager访问端口。例如,在启动时指定端口为9001,启动命令如下:

nohup ./kafka-manager -Dhttp.port=9001 -Dconfig.file=../conf/application.conf &
# 修改Kafka Manager外部访问端口号为9001

(4)关闭Kafka Manager。Kafka Manager没有提供关闭操作的执行脚本及命令,当希望关闭Kafka Manager时,可直接通过kill命令强制杀掉Kafka Manager进程。

查看Kafka Manager进程,输入jps命令,输出以下进程信息:

767 ProdServerStart
12422 QuorumPeerMain
13348 Kafka
895 Jps

其中ProdServerStart即为Kafka Manager进程。通过kill命令关闭Kafka Manager:

kill -9 767             # 关闭Kafka Manager进程

同时,由于Kafka Manager运行时有一个类似锁的文件RUNNING_PID,位于Kafka Manager安装路径bin同目录下,为了不影响下次启动,在执行kill命令后同时删除RUNNING_PID文件,命令如下:

rm –f  RUNNING_PID      # 删除Kafka Manager运行时的PID文件

否则,在下次启动时会由于以下错误而导致Kafka Manager无法启动。错误信息如下:

This application is already running (Or delete /usr/local/software/kafka-manager/ 
RUNNING_PID file).

若想在Kafka Manager监控中能展示更多的信息,则在Kafka启动时启动JMX。至此,Kafka Manager安装讲解完毕,对于Kafka Manager的相关操作将在5.8节进行介绍。

2.6 Kafka源码编译

要研究Kafka,阅读Kafka源码是必不可少的环节。因此,本节将介绍Kafka源码编译及将编译后的源码导入Eclipse的具体步骤。当然也可以将Kafka源码导入其他IDE(如Intellij Idea、STS等)中,大家选用自己惯用的IDE即可。源码导入步骤与导入Eclipse操作基本类似,本书不再做详细介绍。这里只讲解在Windows操作系统下Kafka源码的编译,在其他操作系统上对Kafka源码的编译操作基本类似,只不过添加环境变量操作有所不同,这里不做讲解,读者可以查阅相关资料进行了解。由于Kafka核心模块是用Scala语言开发,用Gradle编译和构建的,因此下面先介绍相关环境的安装配置。

2.6.1 Scala安装配置

由于0.10.1.1版本的Kafka需要Scala版本在2.10以上,因此这里选择scala-2.11.8版本进行安装。

(1)下载并安装。先进入Scala官方网站http://www.scala-lang.org/download/下载相应的安装包,下载图2-4所示版本的Scala。

图2-4 Scala安装包下载列表

下载完成后,直接将安装包解压到指定目录即完成安装,安装时解压到D:\software\scala- 2.11.8目录下。

(2)环境变量配置。安装完成后,配置Scala运行环境变量,在系统变量中新增Scala安装路径配置,编辑系统变量配置如图2-5所示。

然后将;%SCALA_HOME%\bin添加到用户变量path中。与JDK环境安装配置一样,直接添加至自定义的用户环境变量CLASSPATH中,如图2-6所示。

图2-5 新建SCALA_HOME变量指定Scala安装路径

图2-6 修改CLASSPATH添加Scala环境变量

(3)验证。Scala安装及环境变量配置完成后,在Windows下打开一个cmd命令行终端。输入查看Scala版本信息的命令:

scala  -version      # 查询Scala版本

若输出以下信息则表示Scala安装配置成功:

Scala code runner version 2.11.8 -- Copyright 2002-2016, LAMP/EPFL

2.6.2 Gradle安装配置

进入Gradle官方网站https://gradle.org/releases/下载Gradle安装包。本书编写时Gradle的最新版本为gradle-3.3,这里下载的就是这个版本,读者可以根据自己需要选择不同版本进行下载。下载后将Gradle文件解压到相应目录,这里将Gradle解压到D:\software\gradle-3.3目录下,安装及环境变量配置与Scala操作一样,新增系统环境GRADLE_HOME,指定gradle安装路径,并将;%GRADLE_HOME%\bin添加到path中,这里依然是添加到CLASSPATH之中。

Gradle安装及环境变量配置完成之后,打开Windows的cmd命令窗口,输入gradle –version,若输出如图2-7所示信息,则表示Gradle安装配置成功。

图2-7 Gradle安装验证结果

2.6.3 Kafka源码编译

先进入http://kafka.apache.org/downloads.html下载Kafka源码文件。本书编写时Kafka的最新版本为kafka-0.10.1.1,这里我们下载的是kafka-0.10.1.1-src.tgz,将下载的源码包放在F:\kafka-0.10.1.1目录下,解压后如图2-8所示。

图2-8 Kafka源码解压后的文件目录

进入kafka-0.10.1.1-src,Kafka源码包括图2-9所示的目录及文件。

图2-9 Kafka源码包括的目录及文件

Kafka源码对应目录及文件说明如表2-4所示。

表2-4 Kafka源码对应目录及文件说明

名  称

描  述

bin

包括Windows和Linux平台下Kafka相关操作的执行脚本,如启动和关闭KafkaServer、创建主题、分区管理、模拟生产者和消费者基本操作的脚本等

clients

Kafka客户端,包括KafkaProducer和KafkaConsumer,用Java语言开发

config

Kafka运行相关配置文件,如在启动代理时需要加载的server.properties文件

connect

0.9版本之后新增加的特性,提供了Kafka与其他系统整合进行数据导入、导出操作的统一接口,为Kafka能够与其他系统整合构建可水平扩展、高可靠的数据流处理平台提供了一个简单模型,用Java语言开发

core

Kafka的核心代码,包括消息协议定义、日志管理、各组件之间通信、安全协议等

docs

Kafka官方网站相关文档

examples

Kafka实例代码

streams

Kafka 0.10版本之后增加的新特性,是一个用来构建流处理程序的库,用Java语言开发

tools

Kafka提供的工具类,用于查看生产者性能、吞吐量等

tests

系统测试脚本

由于在Kafka源代码的gradle子目录中没有wrapper类库,因此在Kafka根目录下执行gradlew eclipse命令时会报图2-10所示的错误。

图2-10 Kafka源码编译出错信息

接下来安装wrapper类库。由于本地安装的Scala版本为2.11.8,在安装wrapper类库之前,先修改Kafka源码目录下的gradle.properties文件,将Scala版本设置为2.11.8。gradle.properties文件内容如图2-11所示。

图2-11 gradle.properties文件内容

然后进入Kafka源码根目录下,执行gradle wrapper命令来下载wrapper包,如图2-12所示。

图2-12 wrapper安装过程输出信息

在该命令执行过程中会下载相应的jar文件,待完成相应文件下载后,若在控制台打印输出“BUILD SUCCESSFUL”字样则表示安装wrapper类库成功。执行成功后会在Kafka源码的gradle目录下生成wrapper目录,如图2-13所示。

图2-13 wrapper安装过程创建的wrapper目录

进入wrapper目录,在该目录下已创建了一个gradle-wrapper.jar文件,如图2-14所示。

图2-14 wrapper安装过程生成的文件

最后在Kafka源码根目录执行gradlew eclipse命令,对Kafka源码进行编译。这个过程由于要下载一系列依赖包,因此有些耗时,若出现“BUILD SUCCESSFUL”字样,则表示编译完成,如图2-15所示。

图2-15 Kafka源码成功编译输出日志信息

若读者在编译时输入gradlew eclipse命令后控制台打印日志输出:

Downloading https://services.gradle.org/distributions/gradle-3.3-bin.zip  
.........................

一直卡在下载gradle-3.3-bin.zip时,可通过下载工具先下载gradle-3.3-bin.zip文件,然后复制到C:\Users\用户名.gradle\wrapper\dists\gradle-3.3-bin\37qejo6a26ua35lyn7h1u9v2n目录下,接着再次运行gradlew eclipse命令进行编译。

2.6.4 Kafka导入Eclipse

通过前面的步骤已完成了Kafka源码的编译,现在介绍如何将Kafka源码导入Eclipse。在Eclipse视图中选择“import”,在弹出对话框中选择“Existing Projects into Workspace”,指定Kafka源码路径,依次导入Kafka源码中的core和client工程。导入项目后,若Eclipse的编码方式不是UTF-8,会有错误提示,读者在导入Kafka源码时要确保Eclipse已设置workspase的编码方式为UTF-8,同时建议修改Scala使用的JVM版本为1.8,如图2-16所示。

Eclipse工作空间环境配置完毕后,导入Kafka的core和client工程,如图2-17所示。

图2-16 Eclipse设置工程Scala运行的JVM版本界面

图2-17 Kafka源码导入Eclipse效果

若在Eclipse中看到core工程有错误提示信息,则在core工程上右键配置“build path”,在Libraries视图下可以看到缺失如图2-18所示的两个文件,这两个文件都是core工程测试代码所依赖的文件,并不影响core工程本身的运行。这里为了简单,直接将这两个文件从Libraries中移除。

图2-18 Kafka core报错所缺失的文件

若直接运行core工程,kafka.kafka.scala会报出如图2-19所示的错误信息。

图2-19 Eclipse启动Kafka时在控制台输出的错误信息

图2-19所示的错误是由于Kafka启动时需要加载server.properties文件,用于初始化KafkaServer,因此在运行kafka.kafka.scala启动KafkaServer时,需要指定一个配置文件。KafkaServer初始化的配置这里暂不进行详细介绍,将穿插在第3章至第6章对Kafka相关知识的讲解中进行介绍。现在,在Eclipse中设置运行参数,指定server.properties文件路径,配置如图2-20所示。由于Kafka依赖ZooKeeper,因此要保证在启动KafkaServer之前先启动ZooKeeper。

图2-20 Eclipse设置Kafka启动加载配置文件界面

为了在控制台输出启动日志,需要将Kafka源码config目录下的log4j.properties文件复制到Eclipse core工程src/main/scala目录下,运行kafka.scala启动KafkaServer,Eclipse控制台输出启动日志信息如图2-21所示。

图2-21 Eclipse启动KafkaServer输出结果

图2-21所示的日志信息表明:Kafka源码已成功在Eclipse中运行起来。接下来就可以调试Kafka,深入了解Kafka运行机制了。

2.7 小结

本章详细讲解了Kafka运行环境安装部署的步骤,包括在Windows操作系统、Linux操作系统安装部署Kafka,以及Kafka可视化管理工具的安装和Kafka源码的编译等。

目录

相关技术

推荐用户