跳转至

17. kafka安装

1. 下载kafka二进制包

# 二进制包，也就是我们所说的绿色版，解压就能用！！！
# https://kafka.apache.org/
kafka_2.12-3.0.0.tgz

2. 解压缩

tar -zxvf kafka_2.12-3.0.0.tgz

3. 移动到/usr/local目录下

mv kafka_2.12-3.0.0 /usr/local/kafka

4. 查看配置文件

egrep -v "^#|^$" /usr/local/kafka/config/server.properties

broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

配置文件注释

##########注释
##每一个 broker 在集群中的唯一标示，要求是正数。在改变 IP 地址，不改变 broker.id 的话不会影响 consumers
broker.id=0
listeners=PLAINTEXT://192.168.10.40:9092
## broker 处理消息的最大线程数，一般情况下不需要去修改
num.network.threads=3
## broker 处理磁盘 IO 的线程数，数值应该大于你的硬盘数
num.io.threads=8
## socket 的发送缓冲区，socket 的调优参数 SO_SNDBUFF
socket.send.buffer.bytes=102400
## socket 的接受缓冲区，socket 的调优参数 SO_RCVBUFF
socket.receive.buffer.bytes=102400
## socket 请求的最大数值，防止 serverOOM，message.max.bytes 必然要小于 socket.request.max.bytes，
会被 topic 创建时的指定参数覆盖
socket.request.max.bytes=104857600
##kafka 数据的存放地址，多个地址的话用逗号分割 /tmp/kafka-logs-1，/tmp/kafka-logs-2
log.dirs=/tmp/kafka-logs
## 每个 topic 的分区个数，若是在 topic 创建时候没有指定的话会被 topic 创建时的指定参数覆盖
num.partitions=1
# 每个数据目录中的线程数，用于在启动时日志恢复，并在关闭时刷新。
num.recovery.threads.per.data.dir=1
##offset topic 复制因子（ps：就是备份数，设置的越高来确保可用性）。为了确保 offset topic 有效的复制因
27
28
子，第一次请求 offset topic 时，活的 broker 的数量必须最少最少是配置的复制因子数。如果不是，offset topic
将创建失败或获取最小的复制因子（活着的 broker，复制因子的配置）
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
## 数据存储的最大时间超过这个时间会根据 log.cleanup.policy 设置的策略处理数据，也就是消费端能够多久去
消费数据
log.retention.hours=168
## topic 的分区是以一堆 segment 文件存储的，这个控制每个 segment 的大小，会被 topic 创建时的指定参数覆
盖
log.segment.bytes=1073741824
## 文件大小检查的周期时间，是否处罚 log.cleanup.policy 中设置的策略
log.retention.check.interval.ms=300000
#zookeeper 集群的地址，可以是多个，多个之间用逗号分割
hostname1:port1,hostname2:port2,hostname3:port3
zookeeper.connect=192.168.10.40:2181
## ZooKeeper 的连接超时时间
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

5. 修改配置文件

cd /usr/local/kafka/config
ls
cp server.properties server.properties.bak
ls
vim server.properties
# 将第31行的注释打开！

 23 ############################# Socket Server Settings #############################
 24
 25 # The address the socket server listens on. It will get the value returned from
 26 # java.net.InetAddress.getCanonicalHostName() if not configured.
 27 #   FORMAT:
 28 #     listeners = listener_name://host_name:port
 29 #   EXAMPLE:
 30 #     listeners = PLAINTEXT://your.host.name:9092
 31 listeners=PLAINTEXT://:9092

6. 启动服务

#内存可能不够！调整内存到5G

#1.先启动zookeeper
/usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties &

#2.在启动kafka
/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

#3.验证启动
jobs

#4.查看端口9092,2181
netstat -tunlp

[root@node-1 kafka]# jobs
[1]-  Running                 /usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties &
[2]+  Running                 /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
最后更新: 2022-02-20 08:44:07