部署Kafka
#官方文档
http://kafka.apache.org/quickstart
1、环境准备
#在三个节点提前部署jdk和zookeeper
[root@node1 ~]#java -version
openjdk version "1.8.0_342"
[root@node1 ~]#zkServer.sh version
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Apache ZooKeeper, version 3.7.1 2022-05-07 06:45 UTC
#确保三个节点的zookeeper启动
[root@node1 ~]#systemctl start zookeeper
[root@node1 ~]#systemctl status zookeeper.service
2、下载kafka解压
官网：http://kafka.apache.org/downloads
清华源：https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/
#所有节点
[root@node1 ~]#tar xf kafka_2.13-3.0.0.tgz -C /usr/local
[root@node1 ~]#cd /usr/local
[root@node1 local]#ln -sv kafka_2.13-3.0.0 kafka
[root@node1 local]#echo 'PATH=/bin/local/kafka/bin:$PATH' > /etc/profile.d/kafka.sh
[root@node1 local]#. /etc/profile.d/kafka.sh
3、修改配置文件
#第一个节点：
[root@node1 local]#vim /usr/local/kafka/config/server.properties
broker.id=1 # 每个broker在集群中每个节点的正整数唯一标识，此值保存在log.dirs下的meta.properties文件
listeners=PLAINTEXT://10.0.0.101:9092 # 指定当前主机的IP做为监听地址,注意:不支持0.0.0.0
log.dirs=/usr/local/kafka/data # kakfa用于保存数据的目录，所有的消息都会存储在该目录当中
num.partitions=3 # 设置创建新的topic时默认分区数量,建议和kafka的节点数量一致
log.retention.hours=168 # 设置kafka中消息保留时间，默认为168小时即7天
zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 # 指定连接的zk的地址,zk中存储了broker的元数据信息
zookeeper.connection.timeout.ms=6000 # 设置连接zookeeper的超时时间，单位为ms
[root@node1 local]#mkdir /usr/local/kafka/data #数据目录
[root@node1 local]#scp /usr/local/kafka/config/server.properties 10.0.0.102:/usr/local/kafka/config/server.properties
[root@node1 local]#scp /usr/local/kafka/config/server.properties 10.0.0.103:/usr/local/kafka/config/server.properties
=============================================================================
#@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@ 其他配置说明 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
num.network.threads=3 # 处理网络请求的线程数量，默认为3个
num.io.threads=8 # 执行磁盘IO操作的线程数量，默认为8个
socket.send.buffer.bytes=102400 # socket服务发送数据的缓冲区大小，默认100KB
socket.receive.buffer.bytes=102400 # socket服务接受数据的缓冲区大小，默认100KB
socket.request.max.bytes=104857600 # socket服务所能接受的一个请求的最大大小，默认为100M
default.replication.factor=3 # 设置副本数量为3,当Leader的Replication故障，会进行故障自动转移。
num.recovery.threads.per.data.dir=1 # 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
log.flush.interval.messages=10000 # 消息刷新到磁盘中的消息条数阈值
log.flush.interval.ms=1000 # 消息刷新到磁盘中的最大时间间隔,1s
log.retention.bytes=1073741824 # 日志保留大小，超出大小会自动删除，默认为1G
log.segment.bytes=1073741824 # 日志分片策略，单个日志文件的大小最大为1G，超出后则创建一个新的日志文件
log.retention.check.interval.ms=300000 # 每隔多长时间检测数据是否达到删除条件,300s
#第二个节点：
[root@node2 local]#vim /usr/local/kafka/config/server.properties
broker.id=2
listeners=PLAINTEXT://10.0.0.102:9092
#第三个节点：
[root@node3 local]#vim /usr/local/kafka/config/server.properties
broker.id=3
listeners=PLAINTEXT://10.0.0.103:9092
4、启动服务
[root@node1 local]#kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
#@注：vim /usr/local/kafka/bin/kafka-server-start.sh 里的" -Xmx1G-Xms1G" 可以调整内存
# 确保服务启动状态
[root@node1 local]#ss -ntlp | grep 9092
LISTEN 0 50 [::ffff:10.0.0.101]:9092 *:* users:(("java",pid=46161,fd=142))
# 打开zooinspector可以看到三个id
Broker 依赖于 Zookeeper，每个Broker 的id 和 Topic、Partition这些元数据信息都会写入Zookeeper 的 ZNode 节点中
consumer 依赖于Zookeeper，Consumer 在消费消息时，每消费完一条消息，会将产生的offset保存到 Zookeeper 中，下次消费在当前offset往后继续消费.kafka0.9 之前Consumer 的offset 存储在 Zookeeper 中，kafka0.9 以后offset存储在本地
Partition 依赖于 Zookeeper，Partition 完成Replication 备份后，选举出一个Leader，这个是依托于 Zookeeper 的选举机制实现的
5、使用service文件启动
[root@node1 local]#vim /lib/systemd/system/kafka.service
[Unit]
Description=Apache kafka
After=network.target
[Service]
Type=simple
PIDFile=/usr/local/kafka/kafka.pid
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
ExecStop=/bin/kill -TERM ${MAINPID}
Restart=always
RestartSec=20
[Install]
WantedBy=multi-user.target
[root@node1 local]#systemctl daemon-reload
[root@node1 local]#systemctl start kafka.service
[root@node1 local]#systemctl status kafka