Kafka 安装部署、集群启动、命令行操作 与 可视化工具 Kafka Tool

Kafaka 官网：Apache KafkaKafka 官网下载地址：Apache KafkaKafka 官网文档：https://kafka.apache.org/documentation.html#gettingStarted  （！强烈推荐多参考官网！）Github 开源地址：https://github.com/apache/kafkaGithub Kafka 下载地址：Releases · apache/kafka · GitHub

Kafaka 核心能力

1）高吞吐量：使用延迟低至 2 毫秒的计算机群集，以网络有限的吞吐量传递消息。

可扩展：将生产集群扩展到1000个代理、每天数万亿条消息、PB级数据、数十万个分区。弹性扩展和压缩存储和处理。

2）永久储存：在分布式、持久、容错的集群中安全地存储数据流。

3）高可用性：在可用区域上有效地扩展集群，或者跨地理区域连接单独的集群。

Kafka 生态系统

1）内置流处理：通过连接、聚合、筛选器、转换等方式处理事件流，使用事件时间和精确一次处理。

2）接口丰富：Kafka 的现成连接接口集成了数百个事件源和事件接收器，包括Postgres、JMS、Elasticsearch、AWS S3等.

3）客户端库：用大量的编程语言读、写和处理事件流。

4）大型生态系统开源工具：大型开源工具生态系统：利用大量社区驱动的工具。

Kafka 易用性

1）关键任务：支持任务关键型用例，保证排序、零消息丢失和高效的一次性处理。

2）使用广泛：从互联网巨头到汽车制造商再到证券交易所，成千上万的机构都在使用kakfa。超过500万次独特的终身下载.

3）庞大的用户社区：kafka 是阿帕奇软件基金会最活跃的五个项目之一，在世界各地有数百个会议。

4）丰富的在线资源：丰富的文档、在线培训、指导教程、视频、示例项目、堆栈溢出等。

Producer API生产者：允许应用程序将记录流发布到一个或多个 Kafka 主题。Consumer API

消费者：I允许应用程序订阅一个或多个主题，并处理为它们生成的记录流。

Streams API允许应用程序充当流处理器，使用来自一个或多个主题的输入流并生成一个或多个输出主题的输出流，从而有效地将输入流转换为输出流。Connector API连接器：允许构建和运行可重用的生产者或消费者，将Kafka主题连接到现有的应用程序或数据系统。例如，到关系数据库的连接器可能会捕获对表的每个更改。Admin API管理API允许管理和检查主题、代理和其他 Kafka 对象。

https://mirror.bit.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz
https://mirrors.bfsu.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz
https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz

broker.id=0

#broker 的全局唯一id值，用于集群下每个 kafka 的唯一标识，可以是任意的整数值，默认为 0。delete.topic.enable=true

删除 topic 时是否物理删除。默认为 false 或者无此配置项(此时手动添加即可)

1、如果没有配置 delete.topic.enable，或者值为 false，则删除 topic 时是标记删除，不是真正的物理删除，在 log.dirs 配置的目录下仍然能看到数据，以及在 zk 服务器的 /brokers/topics 路径也能看到数据。

2、想要删除 topic 时真正的物理删除，此必须配置为 true.

num.network.threads=3

#处理网络请求与响应的线程数量，默认为 3

num.io.threads=8

#服务器用于处理请求的线程数，可能包括磁盘I/O，默认为 8

socket.send.buffer.bytes=102400

#发送套接字的缓冲区大小，默认为 102400，100 kb

socket.receive.buffer.bytes=102400

#接收套接字的缓冲区大小，默认为 102400，100 kb

socket.request.max.bytes=104857600

#请求套接字的缓冲区大小，默认为 104857600，100M

log.dirs=/home/kafka_2.13-2.6.0/logs

#kafka 运行日志存放的路径，改成自定义的即可，kafka 以日志的形式存储数据，这个路径不能随意删除。

num.partitions=1

#topic在当前broker上的分区个数，默认为 1

num.recovery.threads.per.data.dir=1

#用来恢复和清理data下数据的线程数量，默认为 1

log.retention.hours=168

#日志文件保留的最长时间，超时将被删除，默认 168 小时，7 天

log.segment.bytes=1073741824#日志文件最大大小，超过时会新建log.retention.bytes=1073741824#基于大小的日志（数据）保留策略，当存储的数据量超过此大小时，就删除旧数据。默认为 1G。此配置默认是被注释的。log.retention.check.interval.ms=300000#检查日志段以查看是否可以根据保留策略删除日志段的间隔zookeeper.connect=193.234.2.100:2181,193.234.2.101:2181

#配置连接 Zookeeper 集群地址，默认为 localhost:2181

zookeeper.connection.timeout.ms=18000连接到zookeeper的超时（毫秒），默认 18 秒

[root@localhost ~]# vim /etc/profile                     

export KAFKA_HOME=/home/kafka_2.13-2.6.0
export PATH=$PATH:$KAFKA_HOME/bin

[root@localhost ~]# source /etc/profile                 //刷新配置文件，使用配置生效

[root@localhost ~]# echo $KAFKA_HOME          //获取环境变量的值，查看 kafka 环境变量配置是否配置成功

[root@wangmaoxiong001 bin]# https://blog.csdn.net/wangmx1993328/article/details/kafka-server-start.sh https://blog.csdn.net/wangmx1993328/article/config/server.properties前台启动 kafka 服务器，需要指定配置文件[root@wangmaoxiong001 bin]# nohup  https://blog.csdn.net/wangmx1993328/article/details/kafka-server-start.sh https://blog.csdn.net/wangmx1993328/article/config/server.properties &后台启动 kafka 服务器，需要指定配置文件。此时启动日志在当前目录下的 nohup.out 文件中。[root@wangmaoxiong001 config]# jps          #启动之后可以看到 kafka 进程
4144 Kafka[root@wangmaoxiong001 bin]# https://blog.csdn.net/wangmx1993328/article/details/kafka-server-stop.sh关闭启动 kafka 服务器

1）修改 %KAFKA_HOME/conf% 目录下的 zookeeper.properties 配置文件，修改zk数据存储目录，如： dataDir=/home/kafka_2.13-2.6.0/data/zk-data，zk 服务器默认数据存储目录为：#dataDir=/tmp/zookeeper。zk 服务器默认监听客户端端口为 clientPort=2181。

2）通过 %KAFKA_HOME/bin% 目录下的：

zookeeper-server-start.sh 脚本：启动zk服务器,指定配置文件,如后台启动:nohup https://blog.csdn.net/wangmx1993328/article/details/bin/zookeeper-server-start.sh https://blog.csdn.net/wangmx1993328/article/details/config/zookeeper.properties &。此时启动日志在当前目录下的 nohup.out 文件中。

zookeeper-server-stop.sh 脚本：关闭 zk 服务器

zookeeper-shell.sh 脚本：连接 zk 服务器，比如 zookeeper-shell.sh 127.0.0.1:2181，连接上之后就可以使用 zk 命令行，比如： ls /

Tip1：命令行的所有参数都是两个横杠开头

Tip2：kafka 默认使用 9092 端口进行消息传送，所以集群下的每个 kafka 服务器所在的机器都需要开启 9092 访问端口，否则发送消息、接收消息会失败。

firewall-cmd --zone=public --add-port=9092/tcp --permanent        #开启 9092 端口

firewall-cmd --reload   #重启防火墙

firewall-cmd --zone=public --list-ports  #查看对外成功开放的端口

Tip3：如下所有脚本参数都可以通过 https://blog.csdn.net/wangmx1993328/article/details/xxx.sh --help 来进行查看

https://blog.csdn.net/wangmx1993328/article/details/kafka-topics.sh --list --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092

查看当前服务器中的所有 topic（主题）

--list：表示查询

--bootstrap-server ：kafka 集群地址，格式 host1:port1,host2:port2...，也可以只写单个地址，多个地址的目的是防止服务器故障。

https://blog.csdn.net/wangmx1993328/article/details/kafka-topics.sh --describe --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 -topic helloWorld

查看指定 topic 详细信息

1、比如分区个数，自己所属分区，副本个数，领导者机器id等等

https://blog.csdn.net/wangmx1993328/article/details/kafka-topics.sh --create --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 --replication-factor 2 --partitions 2 --topic helloWorld

创建 topic。

--replication-factor：指定副本数，不能大于集群下的 kafka 节点数，通常与节点数一致即可，这样就保证每个kafka上都复制有一份相同的数据，防止宕机。

--partitions：定义分区数，topic 的分区是提供 kafka 吞吐量的方式之一，数量没有强制要求，比如执行左侧命令后在 log.dirs 目录下可以看到 helloWorld 的两个分区：helloWorld-0，helloWorld-1，当副本数小于 kafka 节点数时，分区也可能分散位于多个 kafka 上。

提醒：创建成功之后，可以在 log.dirs 配置的目录下看到数据，以及在 zk 服务器的 /brokers/topics 路径也能看到数据。

https://blog.csdn.net/wangmx1993328/article/details/kafka-topics.sh --delete --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 --topic helloWorld

删除  topic

提醒：如果 server.properties 没有配置 delete.topic.enable，或者值为 false，则删除 topic 时是标记删除，不是真正的物理删除，在 log.dirs 配置的目录下仍然能看到数据，以及在 zk 服务器的 /brokers/topics 路径也能看到数据。

https://blog.csdn.net/wangmx1993328/article/details/kafka-console-producer.sh --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 --topic helloWorld

启动生产者

--bootstrap-server ：kafka 集群地址，格式 host1:port1,host2:port2...

--topic：表示将消息发送到哪个topic(主题)上，topic 需要已经存在。

提醒1：回车之后就会进入生产者命令行，返回即可发送消息

提醒2：9092 是默认端口，可以自己修改 conf 下的配置文件为需要的端口。

https://blog.csdn.net/wangmx1993328/article/details/kafka-console-consumer.sh --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 --from-beginning --topic hello-world

启动消费者

--bootstrap-server ：kafka 集群地址，格式 host1:port1,host2:port2...，也可以只写单个地址，多个地址的目的是防止服务器故障。

--from-beginning：是否从起始位置接收消息，比如生产者发送消息的时候，消费者还未启动，此时加上此参数就会从起始位置全部接收，否则只会接收消费者自己启动后监听到的数据。根据实际情况决定是否需要读取历史消息。

--topic：接收哪个主题的消息

提醒：回车之后就会进入消费者命令行，此时生产者只要发送消息，这里就会显示。

3.1 Broker Configs ：broker 配置。所以配置项可以通过  server.properties 文件配置

3.2 Topic Configs ：主题配置

3.3 Producer Configs ：生成者配置

3.4 Consumer Configs ：消费者配置

3.5 Kafka Connect Configs ：连接配置

3.6 Kafka Streams Configs ：流配置

3.7 AdminClient Configs ：Kafka管理客户端库的配置。

快速查看您的所有卡夫卡集群，包括他们的经纪人，主题和消费者查看分区中消息的内容并添加新消息查看消费者的补偿，包括阿帕奇风暴卡夫卡消息偏移量以漂亮的打印格式显示 JSON 和 XML 消息添加和删除主题以及其他管理功能将分区中的单个消息保存到本地硬盘编写自己的插件，允许您查看自定义数据格式

Kafka 工具可以运行在 Windows、Linux 和 Mac 操作系统上

kafka Tool 主页：Offset Explorer

Kafka Tool 下载页：Offset Explorer