Kafka安装

一.kafka下载

https://dlcdn.apache.org/kafka/3.1.0/kafka_2.13-3.1.0.tgz

二.解压

cd ~/software

tar ­zxf kafka_2.12­2.3.0.tgz

三.配置

# 创建存放数据的文件夹

cd kafka_2.12­2.3.0

mkdir data

# 修改kafka配置文件

cd config

vi server.properties

listeners=PLAINTEXT://:9092 # kafka默认监听的端口号为9092

log.dirs=../data # 指定数据的存放目录

zookeeper.connect=localhost:2181 # zookeeper的连接信息

四.启动zk

使用kafka内置zk

cd ~/software/kafka_2.12­2.3.0/bin/
./zookeeper­server­start.sh ../config/zookeeper.properties

使用外部zk（推荐）

cd ~/software/zookeeper­3.4.13/bin/
./zkServer.sh start

五.启动kafka

./kafka­server­start.sh ../config/server.properties & 
# &表示后台运行，也可使用daemon选项
jps # 查看进程，jps是jdk提供的用来查看所有java进程的命令

六.创建Topic（主题）

/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic 
hello--partitions 2 --replication-factor 1


# 查看Topic列表
./kafka­topics.sh ­­list ­­bootstrap-server localhost:9092 # __consumer_offsets是
kafka的内部Topic
# 查看某一个具体的Topic
./kafka­topics.sh ­­describe bootstrap-server localhost:90921 ­­topic hello
# 修改Topic：只能增加partition个数，不能减少，且不能修改replication­factor
./kafka­topics.sh ­­alter bootstrap-server localhost:9092 ­­topic hello ­­
partitions 5
# 删除Topic (需要启用topic删除功能)
./kafka­topics.sh ­­delete ­­bootstrap-server localhost:9092 ­­topic hello

七.启动kafka的Producer（生产者 ）

./kafka­console­producer.sh ­­broker­list localhost:9092 ­­topic hello

八.启动kafka的Consumer（消费者）

./kafka­console­consumer.sh ­­bootstrap­server localhost:9092 ­­topic hello ­­
from­beginning

九.配置详解

############################# Server Basics #############################
# broker的id，值为整数，且必须唯一，在一个集群中不能重复
broker.id=0
############################# Socket Server Settings #############################
# kafka默认监听的端口为9092
#listeners=PLAINTEXT://:9092
# 处理网络请求的线程数量，默认为3个
num.network.threads=3
# 执行磁盘IO操作的线程数量，默认为8个
num.io.threads=8
# socket服务发送数据的缓冲区大小，默认100KB
socket.send.buffer.bytes=102400
# socket服务接受数据的缓冲区大小，默认100KB
socket.receive.buffer.bytes=102400
# socket服务所能接受的一个请求的最大大小，默认为100M
socket.request.max.bytes=104857600
############################# Log Basics #############################
# kafka存储消息数据的目录
log.dirs=../data
# 每个topic默认的partition数量
num.partitions=1
# 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
num.recovery.threads.per.data.dir=1
############################# Log Flush Policy #############################
# 消息刷新到磁盘中的消息条数阈值
#log.flush.interval.messages=10000
# 消息刷新到磁盘中的最大时间间隔
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# 日志保留小时数，超时会自动删除，默认为7天
log.retention.hours=168
# 日志保留大小，超出大小会自动删除，默认为1G
#log.retention.bytes=1073741824
# 日志分片策略，单个日志文件的大小最大为1G，超出后则创建一个新的日志文件
log.segment.bytes=1073741824
# 每隔多长时间检测数据是否达到删除条件
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper连接信息，如果是zookeeper集群，则以逗号隔开
zookeeper.connect=localhost:2181
# 连接zookeeper的超时时间
zookeeper.connection.timeout.ms=6000