Kafka 集群安装及常用命令
搭建 Kafka 服务集群
三台服务器（ns1，ns2，ns3），构成集群；
下载解压
下载地址：http://kafka.apache.org/downloads
$ tar xzvf kafka_2.11-2.4.1.tgz -C /home/hadoop/local/
$ ln -s kafka_2.11-2.4.1 kafka
修改配置
$ vim conf/server.properties
修改 id：
broker.id=1ns1 服务器为 1，ns2 服务器为 2，ns3 服务器为 3
修改存放数据目录：
log.dirs=/home/hadoop/local/kafka/datas修改 Zookeeper 服务器地址：
zookeeper.connect=ns1:2181,ns2:2181,ns3:2181/kafka最后的 /kafka 可以不写，如果不写会将所有数据写到根节点 / 下，写了 /kafka 后就会将数据存到 /kafka 节点下；
配置环境变量
$ sudo vim /etc/profile.d/my_env.sh
KAFKA_HOME=/home/hadoop/local/kafka
PATH=$PATH:$KAFKA_HOME/bin
export KAFKA_HOME PATH $ source /etc/profile
分发配置
分发 Kafka
$ cd /home/hadoop/local/
$ xsync kafka_2.11-2.4.1
$ xsync kafka
分发环境变量
$ sudo xsync /etc/profile.d/my_env.sh
在 ns2、ns3 上执行下：
$ source /etc/profile
ns2、ns3 上修改下 broker.id 为 2 和 3
$ vim /home/hadoop/local/kafka/conf/server.properties
启动服务
启动 Kafka 前一定要先把 Zookeeper 集群启动好；
在 ns1、ns2、ns3 上都执行启动命令：
$ kafka-server-start.sh -daemon /home/hadoop/local/kafka/config/server.properties
检查进程：
$ myjps.sh
========== ns1 JPS ==========
4208 Jps
4115 Kafka
24836 DataNode
25014 NodeManager
24711 NameNode
10666 JobHistoryServer
2571 QuorumPeerMain
========== ns2 JPS ==========
7204 DataNode
7638 ResourceManager
7895 NodeManager
25544 Jps
22878 QuorumPeerMain
25454 Kafka
========== ns3 JPS ==========
11364 NodeManager
10917 DataNode
19638 Jps
11116 SecondaryNameNode
18189 QuorumPeerMain
19551 Kafka 都启动了；
停止服务：
$ kafka-server-stop.sh
集群启停脚本
写一个可以在 ns1 一台服务器上就能启动、停止整个集群的脚本；
$ vim /home/hadoop/bin/kafka.sh
#!/bin/bash
if [ $# -lt 1 ]
then
echo "USAGE: kafka.sh {start|stop}"
exit
fi
case $1 in
start)
for i in ns1 ns2 ns3
do
echo "================ START $i KAFKA ==================="
ssh $i /home/hadoop/local/kafka/bin/kafka-server-start.sh -daemon /home/hadoop/local/kafka/config/server.properties
done
;;
stop)
for i in ns1 ns2 ns3
do
echo "================ STOP $i KAFKA ==================="
ssh $i /home/hadoop/local/kafka/bin/kafka-server-stop.sh
done
;;
*)
echo "USAGE: kafka.sh {start|stop}"
exit
;;
esac $ chmod u+x /home/hadoop/bin/kafka.sh
启动：
$ kafka.sh start
================ START ns1 KAFKA ===================
================ START ns2 KAFKA ===================
================ START ns3 KAFKA =================== 查看：
$ myjps.sh
========== ns1 JPS ==========
24836 DataNode
25014 NodeManager
24711 NameNode
10666 JobHistoryServer
2571 QuorumPeerMain
5259 Jps
5164 Kafka
========== ns2 JPS ==========
26577 Kafka
7204 DataNode
7638 ResourceManager
7895 NodeManager
26652 Jps
22878 QuorumPeerMain
========== ns3 JPS ==========
20659 Jps
11364 NodeManager
10917 DataNode
11116 SecondaryNameNode
18189 QuorumPeerMain
20574 Kafka 关闭：
$ kafka.sh stop
Kafka 命令行操作
1）创建 Topic
$ kafka-topics.sh --create --topic zsoft1 --bootstrap-server ns1:9092 --partitions 2 --replication-factor 3
此命令中的 --bootstrap-server 也可以写成 --zookeeper，指定 zookeeper 地址也可以，写为：
$ kafka-topics.sh --create --topic zsoft1 --zookeeper ns1:2181/kafka --partitions 2 --replication-factor 3
partitions：分区，分区数量可以根据情况设置；
replication-factor：副本，副本数量要
2）查看 Topic 列表
$ kafka-topics.sh --list --bootstrap-server ns1:9092
zsoft1
3）查看详情
查看所有 Topic 的详情
$ kafka-topics.sh --describe --bootstrap-server ns1:9092
Topic: zsoft1 PartitionCount: 2 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: zsoft1 Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: zsoft1 Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1查看某个 Topic 的详情
$ kafka-topics.sh --describe --bootstrap-server ns1:9092 --topic zsoft
Topic: zsoft1 PartitionCount: 2 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: zsoft1 Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: zsoft1 Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1因为我当前只有一个 topic，因此显示的是一样的；
4）删除 Topic
$ kafka-topics.sh --delete --topic zsoft --bootstrap-server ns1:9092
5）Kafka 生产者
在一个 shell 窗口中启动生产者，向 zsoft1 这个 Topic 里面生产数据：
$ kafka-console-producer.sh --topic zsoft1 --broker-list ns1:9092
> a
> b
> c6）Kafka 消费者
在一个 shell 窗口中启动消费者，消费 zsoft1 这个 Topic 里面的数据：
$ kafka-console-consumer.sh --topic zsoft1 --bootstrap-server ns1:9092
a
b
c从第一条开始消费（即打开消费者终端以前进入到 topic 中的消息也消费一次），添加 --from-beginning
$ kafka-console-consumer.sh --topic zsoft1 --bootstrap-server ns1:9092 --from-beginning
7）指定组，消费信息：
在 ns2 的 shell 启动消费者，指定组为 g1
$ kafka-console-consumer.sh --topic zsoft1 --bootstrap-server ns1:9092 --group g1
在 ns3 的 shell 启动消费者，指定组也为 g1
$ kafka-console-consumer.sh --topic zsoft1 --bootstrap-server ns1:9092 --group g1
此时在生产者 console 中输入信息，会交替由 ns2 和 ns3 的消费者 console 消费，消息不会重复消费；
Kafka 压力测试
1）Kafka Producer 压测
（1）测试命令：
$ kafka-producer-perf-test.sh --topic test --record-size 100 --num-records 100000 --throughput -1 --producer-props bootstrap.servers=ns1:9092,ns2:9092,ns3:9092
说明：
- record-size：是一条信息有多大，单位是字节；
- num-records：是总共发送多少条信息；
- throughput：是每秒多少条信息，设成 -1，表示不限流，以最大能力去写，可测出生产者最大吞吐量；
- 测试用的 test topic，要设置成单分区，这样以后要想性能提升时候，增加分区即可；
- test 可以不提前创建，当数据进入时会自动创建没有的 topic，默认创建的是单分区、单副本的 topic；
（2）测试输出：
[2022-08-31 17:38:35,478] WARN [Producer clientId=producer-1] Error while fetching metadata with correlation id 1 : {test=LEADER_NOT_AVAILABLE} (org.apache.k
afka.clients.NetworkClient)
100000 records sent, 73313.782991 records/sec (6.99 MB/sec), 326.70 ms avg latency, 706.00 ms max latency, 306 ms 50th, 665 ms 95th, 697 ms 99th, 705 ms 99.9th. - latency：延时
- 吞吐量为 6.99 MB/sec
- 写入的平均延时为 326.70 ms
- 最大延时：706 ms
- 306 ms 50th, 665 ms 95th, 697 ms 99th, 705 ms 99.9th.： 50%的消息 306ms 发送完，95%的消息 665ms 发送完，99%的消息 697ms 发送完，99.9% 的消息 705ms 发送完；
2）Kafka Consumer 压力测试
$ kafka-consumer-perf-test.sh --broker-list ns1:9092,ns2:9092,ns3:9092 --topic test --fetch-size 10000 --messages 500000 --threads 1
说明：
- --fetch-size：指定每次 fetch 的数据大小；
- --messages：总共要消费的消息个数；
输出：
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
WARNING: Exiting before consuming the expected number of messages: timeout (10000 ms) exceeded. You can use the --timeout option to increase the timeout.
2022-08-31 17:48:22:464, 2022-08-31 17:48:35:540, 9.5367, 0.7293, 100000, 7647.5987, 1661939303609, -1661939290533, -0.0000, -0.0001 - 共消费数据 9.5367MB
- 吞吐量 0.7293 MB/s
- 共消费 100000 条
- 平均每秒消费 7647.5987 条
可以根据这些值判断预计消费量需要启几个消费者进行消费；
Kafka 机器数量计算
Kafka 机器数量 = 2 * （峰值生产速度 * 副本数 / 100）+ 1
Kafka 分区数计算
分区数 = Tt / min(Tp, Tc)
Tp：producer 吞吐量
Tc：consumer 吞吐量
分区数一般设置为：3 ~ 10个