文章目录
- 下载Kafka安装包
- 单机安装
- 集群安装
- Kafka 命令行操作
- 数据日志分离
- windows 下安装 Kafka
下载Kafka安装包
Apache所有项目历史版本镜像地址Apache Kafka下载地址
单机安装
下载地址:http://kafka.apache.org/downloadsStep1: 配置JAVA环境,检验环境:java -versionStep2: 下载并解压kafka
# cd /opt
# wget wget https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz
# tar -zxvf kafka_2.13-2.6.0.tgz
# mv kafka_2.13-2.6.0/ kafka
# cd kafka
Step3: 配置kafka 修改相关配置: vim config/server.properties Broker ID启动以后就不能改了: broker.id=1 取消注释,改成本机IP:listeners=PLAINTEXT://192.168.56.70:9092 num.partitions后面增加2行。 发送到不存在topic自动创建。允许永久删除topic。
num.partitions=1
auto.create.topics.enable=true
delete.topic.enable=true
数据文件目录:log.dirs=/opt/kafka/data
Step4: 启动zookeeper,新版的kafka已内置了一个zookeeper环境,可以直接使用,也可以改成指定已运行的ZK。
# 启动ZK,指定zookeeper配置文件
./bin/zookeeper-server-start.sh ./config/zookeeper.properties &
# 后台启动ZK
./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
启动ZK出现问题: /opt/kafka# /opt/kafka/bin/kafka-run-class.sh: line 318: exec: java: not found问题原因: 是Kafka默认/usr/bin/java路径与我们实际的$JAVA_HOME/bin/java路径不一致导致的。解决方案: 设置一个软连接 ln -s /opt/jdk1.8.0_181/bin/java /usr/bin/java
Step5: 启动Kafka
# 启动kafka,指定kafka配置文件
./bin/kafka-server-start.sh ./config/server.properties &
# 后台启动kafka
./bin/kafka-server-start.sh -daemon ./config/server.properties
Step6: 检查服务
root@yangqian:/opt/kafka# netstat -tunlp|egrep "(2181|9092)"
tcp6 0 0 127.0.0.1:9092 :::* LISTEN 2114/java
tcp6 0 0 :::2181 :::* LISTEN 3098/java
删除kafka全部数据步骤: 1、停止每台机器上的kafka; 2、删除kafka存储目录(server.properties文件log.dirs配置,默认为 /tmp/kafka-logs)全部topic的数据目录; 3、删除zookeeper上与kafka相关的znode节点;除了/zookeeper 4、重启kafka。
集群安装
1、集群规划 准备三台机器:192.168.56.71(zk71)、192.168.56.72(zk72)、192.168.56.73(zk73) 每台机器安装Java和zookeeper
2、解压到指定路径下 本次安装的Kafka为2.1.1版本,注意:压缩包前面的版本号为Scala版本号,而后面的才是Kafka版本号
tar -xzvf kafka_2.11-2.1.1.tgz -C /opt/
mv kafka_2.11-2.1.1 kafka
3、配置环境变量 通过修改/etc/profile文件设置KAFKA_HOME环境变量,并将bin目录增加到PATH环境变量中
sudo vim /etc/profile
添加以下内容:
# Set $KAFKA_HOME
export KAFKA_HOME=/opt/kafka
# Add $KAFKA_HOME/bin to PATH
export PATH=$PATH:$KAFKA_HOME/bin
重新加载profile文件,验证环境变量:
source /etc/profile
echo $KAFKA_HOME
4、修改配置文件config/server.properties 1)设置Broker ID Kafka集群中每个主机即Broker的ID必须不同
# The id of the broker. This must be set to a unique integer for each broker.
broker.id=0
2)设置Kafka日志以及数据存放路径 此路径设置的日志和数据的共同存储路径,即各种Partition的数据也存储在此路径下
log.dirs=/opt/kafka/logs
3)开启管理工具的删除Topic功能 如果不开启开启此功能,那么使用命令行删除Topic时只会是标记删除,实际上并未删除,当再次创建同名Topic时就会报错无法创建
delete.topic.enable=true
4)配置Zookeeper集群 zookeeper.connect=zk71:2181,zk72:2181,zk73:2181
5)具体配置
#broker 的全局唯一编号,不能重复
broker.id=0
#删除 topic 功能使能
delete.topic.enable=true
#处理网络请求的线程数量
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 运行日志存放的路径
log.dirs=/opt/kafka/logs
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment 文件保留的最长时间,超时将被删除
log.retention.hours=168
#配置连接 Zookeeper 集群地址
zookeeper.connect=zk71:2181,zk72:2181,zk73:2181
listeners=PLAINTEXT://192.168.56.71:9092
6)将Kafka分发给其他主机并修改对应的Broker ID即可 分别在 zk72 和 zk73 上修改配置文件/opt/kafka/config/server.properties 中的 broker.id=1、broker.id=2注:broker.id 不得重复
scp -r /opt/kafka zk72:/opt/
scp -r /opt/kafka zk73:/opt/
7)启动集群 依次在每台节点上启动kafka
bin/kafka-server-start.sh -daemon config/server.properties
8)关闭集群 依次在每台节点上关闭kafka
bin/kafka-server-stop.sh stop
Kafka 命令行操作
1、查看当前服务器中的所有 topic
sh bin/kafka-topics.sh --list --zookeeper localhost:2181
2、创建 topic
# 创建一个名为first的topic,只有一个副本,一个分区
sh bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic first
选项说明:--topic 定义 topic 名--replication-factor 定义副本数--partitions 定义分区数
3、删除topic
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic first
4、查看topic详情
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic first
5、启动Producer
# 注意这里要填ip地址,不能填localhost
sh bin/kafka-console-producer.sh --broker-list 192.168.56.70:9092 --topic first
Error while fetching metadata with correlation id : {LEADER_NOT_AVAILABLE} 正确处理姿势【kafka】Error while fetching metadata xxx: {TEST=LEADER_NOT_AVAILABLE}【kafka】报错:advertised.listeners参数的重要性(外部访问局域网kafka)
6、启动Consumer
sh bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.70:9092 --topic first
# --from-beginning:会把主题中以往所有的数据都读取出来。
sh bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.70:9092 --topic first --from-beginning
数据日志分离
上面安装Kafka集群后,创建Topic,然后发送消息,发现数据信息和日志文件都存放在logs文件夹下,现在需要将数据信息和日志分离。
1、停止Kafka集群
2、删除logs文件夹
3、需要删除zookeeper存储的Kafka信息
zkData文件夹下的version-2 是zookeeper存储信息的地方,需要删除,删除之前需要停止zookeeper服务。
4、停止zookeeper集群服务 bin/zkServer.sh stop 5、删除所有机器上zkData下的version-2文件
6、启动zookeeper集群服务 bin/zkServer.sh start
7、修改所有机器Kafka配置 log.dirs=/opt/kafka/data
8、启动Kafka集群这时再次创建Topic和发送消息,所有的数据信息都存在了data文件夹下了。
发送的消息存储在对应Topic文件夹下的00000000000000000000.log文件中。
windows 下安装 Kafka
下载地址:https://archive.apache.org/dist/kafka/2.8.1/kafka_2.13-2.8.1.tgz
首先进入到 \kafka_2.13-2.8.0\bin\windows 目录下。
命令如下:
# 启动 zookeeper 服务
zookeeper-server-start.bat ..\..\config\zookeeper.properties
# 启动 kafka 服务
kafka-server-start.bat ..\..\config\server.properties
# 创建 topic 主题
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafka-test-topic
# 查看 topic 列表
kafka-topics.bat --list --zookeeper localhost:2181
# 创建生产者
kafka-console-producer.bat --broker-list localhost:9092 --topic kafka-test-topic
# 创建消费者
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic kafka-test-topic --from-beginning
参考:全网最通俗易懂的Kafka入门简单封装kafka相关的api滴滴Logi-KafkaManager 一站式Kafka监控与管控平台Kafka系列第三篇