Kafka部署及基本使用
安装部署
1.集群规划
hadoop102: 安装zk和kafka
hadoop103: 安装zk和kafka
hadoop104: 安装zk和kafka
2.集群部署
官方下载地址：
http://kafka.apache.org/downloads.html
producer和consumer是用java写的
broker是用scala写的，scala版本对应截图中的scala2.12和2.13
我们下载2.12对应的压缩包即可
上传服务器后解压：
tar -zxvf kafka_2.12-3.6.0.tgz -C /app/
cd /app
mv kafka_2.12-3.6.0 kafka
cd kafka
进入kafka路径后可以看到如下子目录和文件：
3.配置
配置集群
vim config/server.properties
需要配置如下参数：
找到参数broker.id，这个参数是唯一标识，不能重复；
找到参数log.dirs，这个参数是存储kafka数据的，默认在/tmp中，我们需要改到其他目录，如/opt/module/kafka/data
找到参数zookeeper.connect，配置成zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka
配置环境变量
vim /etc/profile.d/my_env.sh
添加
export KAFKA_HOME=/app/kafka
export PATH=$PATH:$KAFKA_HOME/bin
然后source /etc/profile
其他节点也按此进行配置
4.启动
启动kafka前必须先启动zookeeper
zk启动后启动kafka：
bin/kafka-server-start.sh -daemon config/server.properties
用jps查看启动情况
5.启停脚本
使用脚本进行kafka启停，vim kf.sh：
#!/bin/bash
case $1 in
"start"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------启动 $i Kafka-------"
ssh $i "nohup /app/kafka/bin/kafka-server-start.sh
/app/kafka/config/server.properties 2>&1 &"
done
};;
"stop"){
for i in hadoop102 hadoop103 hadoop104
do
echo " --------停止 $i Kafka-------"
ssh $i "/app/kafka/bin/kafka-server-stop.sh "
done
};;
esac
停止集群：
kf.sh stop
注：
- 当传进来start参数，则进行start相关操作；传进来stop参数，进行stop相关操作；
- 一定要写绝对路径；
- 如果没配主机名和ip地址映射的话，主机名可以写成ip地址；
- 需要配ssh互信;
- 需要授予可执行权限；
- kafka集群停止的时候需要一段时间，不会很快就停;
- 必须在所有kafka全部停止后再关闭zookeeper，如果先关闭zookeeper则kafka就关不掉了，只能通过kill -9杀进程，因为kafka的信息都存在zookeeper中。
操作topic命令
对于每个模块，kafka有对应的脚本
- producer：kafka-console-producer.sh
- broker：kafka-topics.sh
- consumer：kafka-console-consumer.sh
先从kafka-topics.sh开始介绍：
查看用法
bin/kafka-topics.sh
1.核心参数
| 参数 | 描述 |
|---|---|
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号 |
| --topic <String: topic> | 操作的 topic 名称 |
| --create | 创建主题 |
| --delete | 删除主题 |
| --alter | 修改主题 |
| --list | 查看所有主题 |
| --describe | 查看主题详细描述 |
| --partitions <Integer: # of partitions> | 设置分区数 |
| --replication-factor<Integer: replication factor> | 设置分区副本 |
| --config <String: name=value> | 更新系统默认的配置 |
2.基本操作
查看当前服务器中的所有 topic：
bin/kafka-topics.sh --bootstrap-server ip:9092 --list
例：
bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list
我们只要连上了102节点，103、104节点都可以查看的到；
如果需要提高可靠性（万一102挂了），可以写成如下形式：
bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092 --list
企业里一般也采取这种方式
创建 first topic：
bin/kafka-topics.sh --bootstrap-server ip:9092 --create --partitions 1 --replication-factor 3 --topic first
选项说明：
topic 定义 topic 名
replication-factor 定义副本数
partitions 定义分区数
查看 first 主题的详情：
bin/kafka-topics.sh --bootstrap-server ip:9092 --describe --topic first
其中可以看到topic名，有多少分区、多少副本；
此外segment.bytes代表了底层存储一个segment是1个G
分区号为0（指定了1给分区，从0开始）
leader显示了leader副本节点
后面也显示了其他节点
修改分区数（注意：分区数只能增加，不能减少，会报错，因为想象一下多个消费者消费不同分区，到不同位置，没法合并）
bin/kafka-topics.sh --bootstrap-server ip:9092 --alter --topic first --partitions 3
再次查看 first 主题的详情
bin/kafka-topics.sh --bootstrap-server ip:9092 --describe --topic first
无法通过命令行的方式修改副本，后续我们有其他手段修改。
删除 topic
bin/kafka-topics.sh --bootstrap-server ip:9092 --delete --topic first
producer命令行操作
查看操作生产者命令参数
bin/kafka-console-producer.sh
| 参数 | 描述 |
|---|---|
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号 |
| --topic <String: topic> | 操作的 topic 名称 |
发送消息
bin/kafka-console-producer.sh --bootstrap-server ip:9092 --topic first
>hello world
如何确定数据有没有发送到topic？
我们还需创建一个消费者去消费，但是消费者会增量的消费topic里的数据，如果需要消费一开始的数据需要加参数--from-beginning
consumer命令行操作
查看操作消费者命令参数
bin/kafka-console-consumer.sh
| 参数 | 描述 |
|---|---|
| --bootstrap-server <String: server toconnect to> | 连接的 Kafka Broker 主机名称和端口号 |
| --topic <String: topic> | 操作的 topic 名称 |
| --from-beginning | 从头开始消费 |
| --group <String: consumer group id> | 指定消费者组名称 |
消费消息
消费 first 主题中的数据。
bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --topic first
把主题中所有的数据都读取出来（包括历史数据）。
bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --from-beginning --topic first
参考：
https://cloud.tencent.com/developer/article/2218258
