江津区做网站在线建站平台

---

官方下载地址

http://kafka.apache.org/downloads.html

---

上传安装包

解压安装包到指定目录

[root@localhost kafka]# pwd
/root/kafka
[root@localhost kafka]# ll
total 110744
-rw-r--r--. 1 root root 113400977 Oct 25 10:25 kafka_2.12-3.6.0.tgz
[root@localhost kafka]#
[root@localhost kafka]#
[root@localhost kafka]#
[root@localhost kafka]#
[root@localhost kafka]# tar -xvzf kafka_2.12-3.6.0.tgz  -C   ./......
......
......

修改解压包名为kafka

[root@localhost kafka]# mv kafka_2.12-3.6.0 kafka

修改config目录下的配置文件server.propertie

[root@localhost kafka]# cd kafka
[root@localhost kafka]# ls
bin  config  libs  LICENSE  licenses  NOTICE  site-docs
[root@localhost kafka]# cd config/
[root@localhost config]# ls
connect-console-sink.properties    connect-file-source.properties   consumer.properties  server.properties
connect-console-source.properties  connect-log4j.properties         kraft                tools-log4j.properties
connect-distributed.properties     connect-mirror-maker.properties  log4j.properties     trogdor.conf
connect-file-sink.properties       connect-standalone.properties    producer.properties  zookeeper.properties
[root@localhost config]# ll server.properties
-rw-r--r--. 1 root root 6896 Sep 29 12:56 server.properties
[root@localhost config]# #broker的全局唯一编号，不能重复，只能是数字。
broker.id=102
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
log.dirs=/opt/module/kafka/datas
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个topic创建时的副本数，默认时1个副本
offsets.topic.replication.factor=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#每个segment文件的大小，默认最大1G
log.segment.bytes=1073741824
# 检查过期数据的时间，默认5分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
#配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理）
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka

配置环境变量

[root@localhost profile.d]# cat /etc/profile.d/my_env.sh
#KAFKA_HOME
export KAFKA_HOME=/root/kafka/kafka
export PATH=$PATH:$KAFKA_HOME/bin
[root@localhost profile.d]#
[root@localhost profile.d]# source /etc/profile
[root@localhost profile.d]#
[root@localhost profile.d]#
[root@localhost profile.d]#

其他机器同上 - 修改配置文件中的brokerid

分别在hadoop103和hadoop104主机上修改配置文件server.properties中broker.id=103、broker.id=104

注：broker.id不得重复

broker.id=103
.......
.......
.......

broker.id=104
.......
.......
.......

启动集群

先启动zk

[root@localhost conf]# cd /root/kafka/
[root@localhost kafka]# cd kafka
[root@localhost kafka]# ll
total 68
drwxr-xr-x. 3 root root  4096 Sep 29 13:00 bin
drwxr-xr-x. 3 root root  4096 Oct 25 11:27 config
drwxr-xr-x. 2 root root  8192 Oct 25 10:29 libs
-rw-r--r--. 1 root root 14973 Sep 29 12:56 LICENSE
drwxr-xr-x. 2 root root  4096 Sep 29 13:00 licenses
-rw-r--r--. 1 root root 28184 Sep 29 12:56 NOTICE
drwxr-xr-x. 2 root root    44 Sep 29 13:00 site-docs
[root@localhost kafka]# cd bin/[root@localhost bin]# ./zookeeper-server-start.sh
USAGE: ./zookeeper-server-start.sh [-daemon] zookeeper.properties[root@localhost bin]# ./zookeeper-server-start.sh -daemon ../config/zookeeper.properties
[root@localhost bin]#

建议使用单独的zk

然后启动kafka

[root@localhost bin]# export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.382.b05-1.el7_9.x86_64/jre
[root@localhost bin]# ./kafka-server-start.sh -daemon ../config/server.properties
[root@localhost bin]#

---

停止

直接调用 kafka-server-stop.sh 即可

注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。

因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。

---

Kraft 方式部署集群----(不使用zookeeper)

https://huaweicloud.csdn.net/637ee7a2df016f70ae4c93b2.html