部署环境：
Server1:10.0.0.31
Server2:10.0.0.32
Server3:10.0.0.33

一、下载二进制kafka包

apt install -y apenjdk-8-jdk

cd /usr/local/src
wget https://mirrors.bfsu.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz
tar xvf kafka_2.13-2.7.1.tgz
ln -sv /usr/local/src/kafka_2.13-2.7.1 /usr/local/kafka

二、修改配置文件

配置文件说明

broker.id=1 #每个 broker 在集群中的唯一标识，正整数。 
listeners=PLAINTEXT://10.0.0.31:9092 #监听地址 
num.network.threads=8 #网络线程
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.dirs=/usr/local/kafka/kafka-logs #kakfa 用于保存数据的目录，所有的消息 都会存储在该目录当中 
num.partitions=1 #设置创新新的 topic 默认分区数量 ,不超过服务器数量
log.retention.hours=168 #设置 kafka 中消息保留时间，默认为 168 小时即 7 天 
#zookeeper.connect 指定连接的 zk 的地址,zk 中存储了 broker 的元数据信息,格式 如下： 
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
zookeeper.connection.timeout.ms=6000 #设置连接 zookeeper 的超时时间，默 认 6 秒钟

kafka 节点 1

mkdir /usr/local/kafka/data    #用来存放日志的目录

vim /usr/local/kafka/config/server.properties
broker.id=1
listeners=PLAINTEXT://10.0.0.31:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181

kafka 节点 2

mkdir /usr/local/kafka/data    #用来存放日志的目录

vim /usr/local/kafka/config/server.properties
broker.id=2
listeners=PLAINTEXT://10.0.0.32:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181

kafka 节点 3

mkdir /usr/local/kafka/data    #用来存放日志的目录

vim /usr/local/kafka/config/server.properties
broker.id=3
listeners=PLAINTEXT://10.0.0.33:9092
log.dirs=/usr/local/kafka/data
zookeeper.connect=10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181

三、各节点启动 kafka

以守护进程启动，shell退出也仍继续执行

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

查看日志，查看kafka启动情况

ss -ntl #9092端口是否处在监听状态
tail -f /usr/local/kafka/logs/server.log

kafka 1 日志
kafka 2 日志

kafka 3 日志

四、验证 zookeeper 中 kafka 元数据

1、Broker 依赖于 Zookeeper，每个 Broker 的 id 和 Topic、Partition 这些元数据信 息都会写入 Zookeeper 的 ZNode 节点中；
2、Consumer 依赖于 Zookeeper，Consumer 在消费消息时，每消费完一条消息， 会将产生的 offset 保存到 Zookeeper 中，下次消费在当前 offset 往后继续消费； ps：kafka0.9 之前 Consumer 的 offset 存储在 Zookeeper 中，kafka0,9 以后 offset 存储在本地。
3、Partition 依赖于 Zookeeper，Partition 完成 Replication 备份后，选举出一个 Leader， 这个是依托于 Zookeeper 的选举机制实现的；

五、测试 kafka 读写数据

5.1 创建 topic

创建名为 lck ，partitions(分区)为 3，replication(每个分区的副本数/每个 分区的分区因子)为 3 的 topic(主题)
在任意 kafaka 服务器操作

root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181 --partitions 3 --replication-factor 3 --topic lck
Created topic lck.

5.2 验证 topic

状态说明：lck 有三个分区分别为 0、1、2，分区 0 的 leader 是 2（broker.id）， 分区 0 有三个副本，并且状态都为 lsr（ln-sync，表示可以参加选举成为 leader）。

root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181  --topic lck
Topic: lck	PartitionCount: 3	ReplicationFactor: 3	Configs:
Topic: lck	Partition: 0	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
Topic: lck	Partition: 1	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
Topic: lck	Partition: 2	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3

5.3 获取所有topic

root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181
lck

5.4 测试发送消息

root@zk1:~# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.31:9092,10.0.0.32:9092,10.0.0.33:9092 --topic lck
>msg1
>msg2
>msg3

5.5 测试获取消息

可以到任意一台 kafka 服务器测试消息获取，只要有相应的消息获取客户端即可

root@zk1:~# /usr/local/kafka/bin/kafka-consconsumer.sh --topic lck --bootstrap-server 10.0.0.31:9092 --from-beginning
msg2
msg1
msg3

5.6 删除 topic

root@zk1:~# /usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 10.0.0.31:2181,10.0.0.32:2181,10.0.0.33:2181 --topic lck
Topic lck is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.