kafka安装与配置

下载kafka地址：https://kafka.apache.org/downloads

注意：

​ 如果安装kafka时，提示Classpath is empty. Please build the project first e.g. by running ‘./gradlew jar -PscalaVersion=2.12.8’

​ 发现从官网下的是源码。。。

​ 不要下载带src的文件

正确文件名形如：kafka_2.13-2.7.0.tgz

ps:我这里下载的是Scala 2.12的

安装kafka

解压到指定路径

tar -zxvf kafka_2.12-2.7.0.tgz -C /export/servers

给目录赋予权限

sudo chmod -R 777 /export

发送安装包至其他节点

scp -r kafka_2.12-2.7.0 slave1:/export/servers
scp -r kafka_2.12-2.7.0 slave2:/export/servers

启动kafka

– 启动hadoop

start-all.sh

– 启动Zookeeper

zkServer.sh start

进入kafka_2.12-2.7.0目录执行以下命令

./bin/kafka-server-start.sh -daemon ./config/server.properties

测试kafka

以下执行的命令都要进入kafka_2.12-2.7.0目录下执行

创建名为 test 的 Topic。

bin/kafka-topics.sh -zookeeper localhost:2181 --create --partitions 5 --replication-factor 1 --topic test

–partitions：分区数

–replication-factor：副本数

查询 Topic 列表

bin/kafka-topics.sh --list --zookeeper localhost:2181

配置 delete.topic.enable 为 true，这样才能删除 topic

查询 Topic 的信息

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

如果未指定 topic 则输出所有 topic 的信息

消息相关

生产者发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

消费者查询消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --group t1

–from-beginning：表示从头开始接收数据

–group：指定消费者组

查询名为 test 的 Topic 的消息。

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -1

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic test --time -2

–time-1 表示要获取指定 topic 所有分区当前的最大位移（历史总消息数），–time-2 表示获取当前最早位移（被消费的消息数），两个命令的输出结果相减便可得到所有分区当前的消息总数。

输出示例：test:0:3

第一个数字0表示分区，第二个数字3表示偏移量。

关闭kafka

bin/kafka-server-stop.sh config/server.properties

关闭Zookeeper

zkServer.sh stop

关闭hadoop

stop-all.sh