Kafka,ZK集群开发或部署环境搭建及实验

本文属于原创，转载注明出处，欢迎关注微信小程序小白AI博客 微信公众号小白AI或者网站 https://xiaobaiai.net 或者我的CSDN http://blog.csdn.net/freeape

1 前言

Kafka其他特点这里就不罗嗦了，本篇主要是集群开发环境的搭建。具体Kafka相关介绍内容和周边分析可以见另一篇文章《Kafka及周边深度了解》。Kafka是依赖Zookeeper的，单机单服务也需要。Apacke Kafka维护团队开始讨论去除Zookeeper了([2019年11月6日](https://cwiki.apache.org/confluence/display/KAFKA/KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorum))，目前，Kafka使用ZooKeeper来存储分区和代理的元数据，并选择一个Broker作为Kafka控制器，而希望通过删除对ZooKeeper的依赖，将使Kafka能够以一种更具伸缩性和健壮性的方式管理元数据，实现对更多分区的支持，它还将简化Kafka的部署和配置。但是目前我们还是需要Zookeeper。

Zookeeper是由Java编写的，所以需要先安装JDK。

本篇文章主要实现以下内容：

• 单机单Kafka Broker和单Zookeeper(默认安装好后启动ZK和Kafka就是了)
• 单机Kafka Broker集群和Zookeeper集群
• 多机Kafka Broker集群和Zookeeper集群

另，Zookeeper在Kafka中是自带的，这里就不另外安装Zookeeper了，目的只是构建开发环境。安装环境为Ubuntu16.04。

2 安装

2.1 OpenJDK

这里选择OpenJDK 8安装：

# 软件源用的是清华的
$ sudo apt-get update
$ sudo apt-get install openjdk-8-jdk
# 多版本JDK管理和切换，怎么设置更加参考update-alternatives的用法
sudo update-alternatives --config java
# 查看安装后版本(openjdk version 1.8.0_222)
$ java -version

2.2 Kafka版本和安装

2.2.1 版本说明

Kafka也与Hadoop类似，也有着不同的发行版:

• Apache Kafka : 迭代速度快，社区响应高，缺乏高级功能，没有提供任何监控框架或工具，有开源监控框架如Kafka manager，连接器（connector）也比较单一，没有与其他外部系统交互的连接器，需要自行编码；
• Confluent Kafka ： LinkedIn离职员工创办了Confluent，专注于提供基于Kafka的企业级流处理解决方案，比如跨数据中心备份、Schema；Confluent Kafka目前分为免费版和企业版两种，免费版包含Schema注册中心、REST proxy访问Kafka的各种功能、更多连接器，但是没有集群监控、跨数据中心备份等高级功能；
• Cloudera/Hortonworks Kafka : 安装、部署、管理、监控方便，但是降低了kafka的掌控，全由界面操作，而且版本相对社区版滞后；

Kafka1.0和2.0这两个大版本主要还是Kafka Streams的各种改进，在消息引擎方面并未引入太多的重大功能特性。0.11版本在消息引擎处理方面是比较稳定的一个版本。

Apache官网上发布的Kafka版本有如下信息：

因为Kafka是由Scala语言编写的，图中的2.11和2.12指的是Scala的版本，意思就是由Scala2.11或2.12编译出的Kafka2.3.1二进制版本，而2.3.1才是Kafka的版本。

2.2.2 下载安装

从官方网站 https://kafka.apache.org/downloads 下载Kafka，或者从Github上项目Releases下下载，目前为止最新版本为2.3.1(2019年10月24号发布)。这里我们直接下载二进制版本，如果下载源码版本需要自己去编译。

解压后就可以了，里面是支持Linux和Windows启动脚本的。

3 集群及配置

3.1 默认相关配置(单机单Kafka Broker和单ZK)

• Kafka服务默认配置文件./config/server.properties
  • broker id配置
  • 日志文件输出目录/tmp/kafka-logs
  • 每个主题的默认日志分区数为1
  • 相关的线程数配置
  • 相关的IO接收发送缓存大小设置
  • ZK服务地址配置，默认为localhost:2181
  • 日志文件保留的最短时限设置，默认为168小时
  • 服务默认监听端口为9092
  • …
• ZK的默认配置文件./config/zookeeper.properties
  • 快照数据目录/tmp/zookeeper
  • 服务端口2181
  • 每个客户端IP使用的连接数限制为不限(0)
  • …
• 其他配置文件(这里就不展开了，具体使用的时候再来了解)
  • ./config/log4j.properties
  • ./config/connect-log4j.properties
  • ./config/producer.properties
  • ./config/consumer.properties
  • ./config/connect-console-sink.properties
  • ./config/connect-console-source.properties
  • …

启动Zookeeper，守护进程状态运行： ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

启动Kafka，后台运行：./bin/kafka-server-start.sh config/server.properties &

注意：启动Kafka之前是一定要先启动Zookeeper的，如开始前言所说，Kafka是依赖ZK的。

3.2 单机Kafka Broker集群和Zookeeper集群配置

单机Kafka Broker集群和Zookeeper集群配置的实现，只需要启动多个多个Broker和ZK，每个服务设置不同的监听端口就好了，并设置不同的日志目录（这里举例三个broker）：

# Kafka 3个broker配置
# kafka broker1
broker.id=1
listeners = PLAINTEXT://localhost:9092
log.dir=/data/kafka/logs-1
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
# kafka broker2
broker.id=2
listeners = PLAINTEXT://localhost:9093
log.dir=/data/kafka/logs-2
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
# kafka broker3
broker.id=3
listeners = PLAINTEXT://localhost:9094
log.dir=/data/kafka/logs-3
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183

# Zookeeper 3个节点配置
# zk 1
dataDir=/tmp/zookeeper-1
clientPort=2181
maxClientCnxns=0
initLimit=5
syncLimit=2
server.1=localhost:2666:3666
server.2=localhost:2667:3667
server.3=localhost:2668:3668
# zk 2
dataDir=/tmp/zookeeper-2
clientPort=2182
maxClientCnxns=0
initLimit=5
syncLimit=2
server.1=localhost:2666:3666
server.2=localhost:2667:3667
server.3=localhost:2668:3668
# zk 3
dataDir=/tmp/zookeeper-3
clientPort=2183
maxClientCnxns=0
initLimit=5
syncLimit=2
server.1=localhost:2666:3666
server.2=localhost:2667:3667
server.3=localhost:2668:3668

说明：ZK的配置中server.x，这个x是Kafka的Broker ID。而Zookeeper有三个服务，其中一个是主服务，其他两个是从服务，如果有一台服务器关闭，则Zookeeper会自动选择leader。服务器监听三个端口，如上举例中：2181用于客户端连接；2666用于从服务器连接（如果它是领导者）；3666用于leader选举阶段的其他服务器连接;ZooKeeper服务器以两种模式运行：独立和复制模式(或叫仲裁模式，复制模式常用于生产环境)，独立模式就是只有一台服务器，或者说是只有一个服务。此外，复制模式下initLimit是集群中的follower(从)服务器与leader(主)服务器之间初始连接 时能容忍的最多心跳数（tickTime的数量），而tickTime是Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔，默认值3000，也就是每个 tickTime 时间就会发送一个心跳，单位毫秒；syncLimit是集群中的从服务器与主服务器之间请求和应答之间能容忍的最多心跳数（tickTime的数量）。

一一启动ZK服务：

$ ./bin/zookeeper-server-start.sh -daemon config/zookeeper1.properties
$ ./bin/zookeeper-server-start.sh -daemon config/zookeeper2.properties
$ ./bin/zookeeper-server-start.sh -daemon config/zookeeper3.properties

但是会报/tmp/zookeeper-1/myid file is missing异常，这是因为我们现在是单机多服务部署，要想ZK能够识别每一个服务，需要在dataDir这个参数对应目录创建并设置myid文件，里面填一个数字，就是server.x的x这个对应数字。

启动好所有ZK服务后，一一启动Kafka Broker:

$ ./bin/kafka-server-start.sh  config/server1.properties &
$ ./bin/kafka-server-start.sh  config/server2.properties &
$ ./bin/kafka-server-start.sh  config/server3.properties &

3.3 多机Kafka Broker集群和Zookeeper集群配置

多级Kafka Broker集群和Zookeeper集群配置跟单机多Kafka Broker集群 Zookeeper集群是一样的。不过ZK端口和Kafka端口可以设置成一样，注意此时的连接ip都是各个主机的ip地址。如果没有多机，可以用Docker去模拟实现。

4 日志配置

$KAFKA_HOME/bin/kafka-run-class.sh

5 实验

5.1 消息发布订阅

接下来我们用Kafka里提供了脚本工具来测试主题的发布/订阅，只有单机单Broker和单ZK时，--zookeeper、broker-list和bootstrap-server指定单个即可。

1. 使用 kafka-topics.sh 创建单分区单副本的主题users。

# 创建后，主题会持久化到本地，重启服务后还有，需要用--delete选项删除
$ ./bin/kafka-topics.sh --create --zookeeper localhost:2181,localhost:2182,localhost:2183 --replication-factor 1 --partitions 1 --topic users
# 删除主题(主题已经在订阅的过程中是无法删除的)
$ ./bin/kafka-topics.sh --delete --zookeeper localhost:2181,localhost:2182,localhost:2183 --topic users

2. 查看我们创建的主题

$ ./bin/kafka-topics.sh --list --zookeeper localhost:2181,localhost:2182,localhost:2183

3. 往主题users发送消息

$ ./bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093,localhost:9094 --topic users
> hello
> world

4. 接收消息并在终端打印

# --from-beginning 是指将历史未消费消息开始消费（针对同一个消费者，不同的消费者都会从最早的消息开始消费）
$ ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --topic users --from-beginning

然后就可以收到生产者发送的消息了。

注意：Kafka 从 2.2 版本开始将 kafka-console-consumer.sh、kafka-topics.sh等脚本中的 −−zookeeper (用该参数，则消息由ZK管理)参数标注为 “过时”，推荐使用 −−bootstrap-server 参数，−−bootstrap-server指定的不是zookeeper的服务地址，而是Kafka的服务地址，消息由Kafka管理。很显然，zookeeper就会慢慢被取代了。另副本和分区的详细理解可以参考《Kafka及周边深度了解》。

5.2 Kafka Connect导入导出数据

Kafka Connect的作用我们从上图中可以看到，可以从本地数据库或者文件中导入数据，通过Kafka Connect Source到Brokers集群再到Kafka Connect Sink(或者到指定的Topic，图中没有示意出来)，再到消费者或者其他目标数据库。这里我们展示通过本地一个文本文件中写入数据，然后实现上面的数据通路。需要用到connect-standalone.sh或者connect-distributed.sh（Kafka Connect集群）。

1. 配置connect-standalone.sh的配置文件./config/connect-standalone.properties，单Broker可以不用配置

bootstrap.servers=localhost:9092,localhost:9093,localhost:9094

2. 配置connect-file-source.properties参数（没做任何修改，保持默认配置）

# 默认输入是文件流类型，这里主要是配置输入的文件名，和创建的主题
name=local-file-source                                                                                                                      
connector.class=FileStreamSource                                                                                                                             
tasks.max=1                                                                                                                                                  
file=test.txt                                                                                                                                                
topic=connect-test

3. 配置connect-file-sink.properties参数（没做任何修改，保持默认配置）

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1    
file=test.sink.txt                                                                                                                                           topics=connect-test   

4. 本地创建配置中输入源test.txt

echo -e "xiaobaiai.net\nethan" > test.txt

5. 开始导入

$ ./bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

6. 查看结果，本地会生成一个sink配置文件中生成的文件test.sink.txt，另外如果继续往test.txt中输入内容，则这个生成文件里也会接收到内容，或者直接消费connect-test主体，也可以获取内容:

xiaobaiai.net
ethan

$ ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093,localhost:9094 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"xiaobaiai.net"}
{"schema":{"type":"string","optional":false},"payload":"ethan"}

5.3 Kafka Streams实验

具体参见 https://kafka.apache.org/23/documentation/streams/quickstart ，这里暂时不展开了。

6 Kafka其他配置说明

示例：

# ---------------------------
# ./config/server.properties
# ---------------------------
# 默认设置创建主题时副本为1
default.replication.factor=1
# 指定创建主题时默认分区数为3
num.partitions=3

配置项	类型	默认值	示例	描述
broker.id	整型	0	0	kafka broker的id
num.network.threads	整型	3	3	kafka接受和发送消息的线程个数，跟机器cpu核数有关系
num.io.threads	整型	8	8	服务器用于处理IO请求的（可能包括磁盘I/O）的线程数
socket.send.buffer.bytes	整型	102400	102400	linux os的SO_SNDBUF配置参数，如果-1用OS默认配置。
socket.receive.buffer.bytes	整型	102400	102400	linux os的SO_RCVBUF配置参数，如果-1用OS默认配置。
socket.request.max.bytes	整型	104857600	104857600	套接字消息最大多少字节
log.dirs	字符串	/tmp/kafka-logs	/tmp/kafka-logs	kafka的数据目录
num.partitions	整型	1	1	默认partition个数
num.recovery.threads.per.data.dir	整型	1	1	启动时用于日志恢复和关闭时刷新的每个数据目录的线程数
offsets.topic.replication.factor	整型	1	1	自动创建topic的时候，当可用节点个数小于这个数字时候，会创建失败直到有充足的节点可用
transaction.state.log.replication.factor	整型	1	1	transaction topic的复制因子数
transaction.state.log.min.isr	整型	1	1	覆盖min.insync.replicas 的配置
log.retention.hours	整型	168	168	数据最久保存7天
log.segment.bytes	整型	1073741824	1073741824	kafka一个topic有多个partition组成，一个partition一个segment文件存储，当达到log.segment.bytes或者log.roll.hours（log.roll.ms）阈值的时候会新建一个新的segment文件
log.retention.check.interval.ms	整型	300000	300000	log clean检查的interval， 默认5分钟
zookeeper.connect	字符串	localhost:2181	localhost:2181	zookeeper服务连接配置
zookeeper.connection.timeout.ms	整型	6000	6000	zk连接超时设置
group.initial.rebalance.delay.ms	整型	0	0	更多的group JVM消费进程加入进来的时候，距离上次rebalance的时间间隔

更多可以参见：https://kafka.apache.org/documentation/#brokerconfigs

7 扩展

Kafka是一个具有多种代理架构的高可用性服务。一个主题对应于多个分区，一个分区可以有多个副本。这些副本存储在多个代理中以获得高可用性。但是，尽管有多个分区副本集，但只有一个工作的副本集。默认情况下，第一个分配的副本集（首选副本）是负责写入和读取数据的Leader。当我们升级代理或更新代理配置时，我们需要重新启动服务，然后我们需要将分区转移到可用的代理。这里有三种情况:

• 直接关闭Broker：当Broker关闭时，Broker集群将重新选择一个新的Broker作为分区领导，并且Broker上的分区在选举期间将短期不可用
• 打开controlledShutdown：当代理关闭时，代理本身将首先尝试将领导角色转移到其他可用的代理
• 使用命令行工具：使用bin/kafka-preferred-replica-election.sh手动触发分区负责人角色转换

8 总结

本篇是实践的第一环节，实现了Kafka的集群开发环境搭建，并做了主题创建、消息发布、订阅的实验，下一篇将实现Spring Boot集成Kafka，继续！

9 参考资料

• https://github.com/apache/kafka/releases
• https://kafka.apache.org/downloads
• https://medium.com/@kiranps11/kafka-and-zookeeper-multinode-cluster-setup-3511aef4a505
• http://https//segmentfault.com/q/1010000010623287
• https://zookeeper.apache.org/doc/r3.1.2/zookeeperStarted.html
• http://www.corejavaguru.com/bigdata/zookeeper/architecture
• https://zookeeper.apache.org/doc/r3.2.2/zookeeperAdmin.html#id_tickTime
• https://docs.confluent.io/current/zookeeper/deployment.html
• https://spring.io/projects/spring-kafka
• https://kafka.apache.org/quickstart
• https://kafka.apache.org/documentation/#brokerconfigs

本文属于原创，转载注明出处，欢迎关注CSDNfreeape或微信小程序小白AI博客 微信公众号小白AI或者网站 https://xiaobaiai.net