Kafka 服务端安装和配置
本小节主要为安装和运行 Kafka，并使用自带的命令行工具使用 Kafka 最核心的功能–消息的发送和消费。作为 Kafka 笔记的“Hello World”。
这里 Kafka 服务端运行环境为 Linux，需要提前安装和配置好 Linux 和 JDK。如何安装可以参考 Ubuntu Server 安装与设置 https://my.oschina.net/RyenAng/blog/4485236
1、下载 Kafka 和 Zookeeper
Kafka 官网下载 https://kafka.apache.org/downloads 这里下载 Kafka 2.7 版本。需要注意的是，kafka_2.13-2.7.0 中的前面的 2.13 表示编译源码的 Scala 版本，2.7.0 才是 Kafka 的版本号。
Kafka 依赖 Zookeeper 来继续服务的协调和持久存储一些 Kafka 像 topic 主题的元数据信息，虽然 Kafka 已经自带了 Zookeeper，但这里推荐使用单独的 Zookeeper。
Zookeeper 官网下载 https://zookeeper.apache.org/releases.html 下载 Zookeeper 的二进制压缩包，
下载完成后，可以通过 scp 命令或者 FTP 客户端工具上传到 Linux 中。
2、安装单机版的 Zookeeper
Kafka 可以直接运行单机版的 Zookeeper，如果不想单独安装，则可以使用参考 Kafka 单机版安装教程启动 Zookeeper。
# 解压 ZK
tar -xvf apache-zookeeper-3.X.X-bin.tar.gz
mv apache-zookeeper-3.X.X-bin zookeeper
cd zookeeper
配置单机版的 Zookeeper 比较简单，只需要配置 dataDir 指向你存放 ZK 数据的目录即可，因为默认的存储目录会导致重启后，再次运行 Kafka 因为一些元数据不一致而报错。
# zookeeper 配置文件名为 zoo.cfg
cp conf/zoo_sample.cfg conf/zoo.cfg
# 修改 Zookeeper 的配置
vim conf/zoo.cfg
# 找到 dataDir，修改为你存放 zk 数据的目录，例如
dataDir=Zookeeper安装目录下的/data文件夹
这里假设你将 $ZOOKEEPER_HOME 安装目录下的 bin 文件夹添加到了 $PATH 目录。
# 启动 zk, 可以将 bin 目录添加到系统的 PATH，方便直接启动 zk
zkServer.sh start
# 查看 zk 的状态
zkServer.sh status
# 启动 Zookeeper 的命令行客户端
zkCli.sh
# 关闭 zk
zkServer.sh status
集群版的 Zookeeper 这里不做展开，ZK 常用的命令可以参考 https://www.jianshu.com/p/71e647d766c8
3、安装单机版 Kafka
# 解压 Kafka
tar -xvf kafka_2.13-2.7.0.tgz
mv kafka_2.13-2.7.0 kafka
cd kafka
将 Kafka 的安装目录添加到 K A K F A H O M E 变 量 中 ， KAKFA_HOME 变量中，KAKFAHOME变量中，KAKFA_HOME/bin 添加到 $PATH 变量中。
启动 Kafka 自带的 Zookeeper，如果没有安装 Zookeeper 的情况下临时启动一个 Zookeeper 供 Kafka 服务器运行时候使用，不过这里推荐参考上文单独安装一个 Zookeeper 服务。
# 开启 Zookeeper
zookeeper-server-start.sh config/zookeeper.properties
# 关闭命令
zookeeper-server-stop.sh
修改配置文件 $KAFKA_HOME/conf/server.properties，其他的一些配置项说明在文章的最后面。
# Broker 的唯一 id 值
broker.id=0
# 消息存储目录，多个目录使用英文逗号 , 隔开
log.dirs=Kafka安装目录下的/kafka-log-dirs
# 多个目录示例，如果有多个硬盘可以通过此方式将 log 存储在不同的硬盘上
log.dirs=/path1/folder1,/path2/folder2,/path3/folder3
启动和关闭 Kafka， Kafka 的日志输出默认在 $KAFKA_HOME/logs，如果遇到有报错可以通过 log 文件定位问题。启动完成后可以通过 jps 命令查看 Kafka 进程是否成功运行
# 直接启动，要关闭的话 Ctrl + C
kafka-server-start.sh $KAFKA_HOME/config/server.properties
# nohub 启动
nohup kafka-server-start.sh $KAFKA_HOME/config/server.properties >/dev/null &2>&1 &
# 关闭 Kafka 的命令
kafka-server-stop.sh
Kafka 运行成功后，首先通过 Kafka 自带的命令行工具，进行消息的生产和消费，正式开始 Kafka 的 Hello World 之旅。
4、Kafka 的 Hello World
4.1、创建主题
这里新建一个名称为 topic 的主题，需要注意的副本数量要小于等于 Kafka 集群的节点数量，因为此时是单机版的 Kafka，所以副本数量只能设置为 1.
# localhost:9092 Kafka 监听地址
# --replication-factor 副本数
# --partitions 分区数
# --topic 主题名称
kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
创建完成后，会在 Kafka 配置的 log.dirs 目录下新建一个 test-0 的文件夹，其中 test 是 topic 的名称，0 表示第一个分区。
4.2、消息的生产和消费
topic 主题创建后，通过如下命令启动 Kafka 的 Producer 的命令行客户端：
# --topic 指定消息发送的目的 Topic
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
# 启动完成后，会出现一个 > ,在其后输入 Hello Kafka 然后回车
>Hello Kafka
启动 Consumer 消费消息，可以看到你在 Kafka Producer 命令行客户端输入的文字消息。
# --from-beginning 读取全部消息
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
至此，Kafka 的 Hello World 之旅结束，Kafka 消息的生产和消费作为 Kafka 最核心的功能，也是最为我们熟知的功能，本文的后面为如何安装集群版的 Kafka 和一些服务器的配置项说明。
5、安装集群版的 Kafka
这里演示三个服务版的 Kafka 小集群，三个服务器的 hostname 和 IP 的 /etc/hosts 配置示例如下，同时三台服务器之间已经配置好了 ssh 免密登录。
192.168.128.11 node1
192.168.128.12 node2
192.168.128.13 node3
将单机版的 Kafka 安装包使用 scp 拷贝到另外两台的服务器，使用 ssh 登录另外两台机器 node2、node3，分别修改 Kafka 的配置文件 $KAFKA_HOME/config/server.properties，找到如下的配置项，进行修改
# 每个 Broker 的 ID 需要不一样
broker.id=1
# 消息 log 存储目录，可以是一样的，只要有 Kafka 有权限进行写入读取
log.dirs=/home/xian/software/kafka/kafka-log-dirs
# Zookeeper 的地址，用来指向同一个 zk 服务
zookeeper.connect=node1:2181
# 集群版的 Zookeeper 配置参考如下,用英文逗号 , 隔开
zookeeper.connect=node1:2181,node2:2181,node3:2181
在 node2、node3 分别运行启动 Kafka 单机版的命令，至此 Kafka 简单的集群版已经安装完成。
Kafka Broker 的几个配置
Kafka Broker 服务端的配置文件为 $KAFKA_HOME/conf/server.properties，下面列举几个常用的配置说明：
broker.id：集群中每个 Broker 需要不一样的 id
log.dirs：指定消息存储的目录，多个目录用英文逗号隔开
zookeeper.connect：指定 ZK 的地址和端口，多个 ZK 服务用英文逗号隔开
在前面的安装和配置中，可以看到以上三个配置是必须的。
listeners： Kafka Broker bind 的地址，不设置默认为 PLAINTEXT://hostname:9092
PLAINTEXT 协议名，还有 SSL 等协议
hostname:9092, 主机名和端口号
advertised.listeners：Broker 暴露给外部客户端使用的地址，也是写到 zk 服务中的地址，不设置默认为 listeners
num.partitions：默认值 1，指定新建一个主题的时候默认多少个分区
default.replication.factor：默认值 1，指定一个分区默认有多少个副本 Replica。
6、在安装过程中遇到的一些报错
元数据不一致
输出如下，原因是连接到不同的 ZK 服务中。
ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.InconsistentClusterIdException: The Cluster ID 7QbcLfTmRcO2t9u3g5T9LQ doesn't match stored clusterId Some(Jvq4MQIRT9SmyafGNor5GA) in meta.properties. The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.
at kafka.server.KafkaServer.startup(KafkaServer.scala:252)
at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)
at kafka.Kafka$.main(Kafka.scala:82)
at kafka.Kafka.main(Kafka.scala)
解决：删除 log.dir 文件，重启 Kafka，另外，要参考上面的 ZK 安装方式，设置 ZK 数据的存储目录，不然 ZK 重启后还是会发生这种情况。
ZK 依赖的 Jar 有多个
报错输出如下，原因是引入了多个 zk jar 包
java.lang.NoSuchMethodError: org.apache.zookeeper.ZooKeeper.<init>(Ljava/lang/String;ILorg/apache/zookeeper/Watcher;Lorg/apache/zookeeper/client/ZKClientConfig;)V
at kafka.zookeeper.ZooKeeperClient.<init>(ZooKeeperClient.scala:111)
at kafka.zk.KafkaZkClient$.apply(KafkaZkClient.scala:1881)
at kafka.server.KafkaServer.createZkClient$1(KafkaServer.scala:441)
at kafka.server.KafkaServer.initZkClient(KafkaServer.scala:466)
at kafka.server.KafkaServer.startup(KafkaServer.scala:233)
at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)
at kafka.Kafka$.main(Kafka.scala:82)
at kafka.Kafka.main(Kafka.scala)
解决方法是echo $CLASSPATH 后，排除一下可能有 zk jar 包的，像 Hadoop、hbase 等。
Controller 连接不到
报错输出如下：单机版才会输出这个，可以忽略。
[2021-03-05 19:12:16,281] WARN [Controller id=2, targetBrokerId=2] Connection to node 2 (/192.168.128.12:9092) could not be established. Broker may not be available.
解决：listeners 配置要使用 hostname 而不是使用 IP，或者直接不配置 listeners。关于 listener 和 advertised.listeners 的参考：https://www.cnblogs.com/xuliang666/p/11871389.html
小结
本小节中，主要介绍了 Kafka 的安装和运行，并使用自带的命令行工具，生产和消费了第一条消息：Hello Kafka，不过在实际使用中，基本还是通过 Kafka 对应的 API 进行消息的生产和消费。下一小节中，将使用 Java 的 API ，通过示例开发一个 Java 版的“Hello Kafka”！