我的 Kafka 旅程 - Linux下的安装 & 基础命令

• 我的 Kafka 旅程 - 概念 / 模式 / 应用
• 我的 Kafka 旅程 - Linux下的安装 / 基础命令 / 集群
• 我的 Kafka 旅程 - Producer
• 我的 Kafka 旅程 - Broker
• 我的 Kafka 旅程 - 文件存储机制
• 我的 Kafka 旅程 - Consumer
• 我的 Kafka 旅程 - 性能调优 / 压力测试

一、准备工作

1.1 安装解压缩工具 tar

作者：[Sol·wang] - 博客园，原文出处：https://www.cnblogs.com/Sol-wang/p/16691608.html

# 检查是否安装了解压缩工具 tar
tar --version
# 如未安装 tar
yum install tar -y

1.2 安装必备的 java

# 检查是否安装了 java-openjdk，这里选择 java-17-openjdk 版
java -version
# 最低版本java-1.8.0-openjdk，如未安装 java-openjdk
yum install java-17-openjdk -y

二、安装 kafka

官网下载地址：https://kafka.apache.org/downloads，这里下载 kafka_2.13-3.2.3.tgz 版。

# 下载 kafka 安装包文件，下载到当前目录
curl -O https://downloads.apache.org/kafka/3.2.3/kafka_2.13-3.2.3.tgz
# 解压安装包，到指定目录（做为安装目录）
tar -xzvf kafka_2.13-3.2.3.tgz -C /usr/local

三、运行

在 2.8 之前，kafka 依赖于 zookeeper；2.8 之后，zookeeper 的替代品 karft。
以下命令均在解压包(kafka)的根目录进行。

3.1 基于 zookeeper 方式

# 启动
#
# 切换到解压后的 kafka 根目录
cd /usr/local/kafka_2.13-3.2.3
# 启动 zookeeper
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
# 启动 kafka
bin/kafka-server-start.sh -daemon config/server.properties
#
# 验证启动；zookeeper 默认端口2181，kafka 默认端口为9092
ss -plnt			# 端口列是否列出了 2181、9092，表示启动正常
#
# 按顺序 停止 kafka
bin/kafka-server-stop.sh
# 按顺序 停止 zookeeper
bin/zookeeper-server-stop.sh

3.2 基于 kraft 方式

# 启动
#
# 切换到解压后的 kafka 根目录
cd /usr/local/kafka_2.13-3.2.3
#
# 为首次启动，准备 uuid
bin/kafka-storage.sh random-uuid
# 为首次启动，配置 uuid
bin/kafka-storage.sh format -t {uuid} -c config/kraft/server.properties
# 后续的启动，通过 kraft 配置 后台启动 kafka
bin/kafka-server-start.sh -daemon config/kraft/server.properties
#
# 验证启动；kafka 默认端口为 9092
ss -plnt		# 端口列是否列出了 9092，表示启动正常
#
# 停止 kafka
bin/kafka-server-stop.sh

四、基本命令

开放端口：确保每个服务器所应用到的端口(2181 和 9092等)开放（参考firewall）。

4.1 Topic

# 查看已有的 Topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
# 创建一个新的 Topic，分区数1个，副本数3份
bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic {name} --create \
--partitions 1 --replication-factor 1
# 查看一个 Topic 详细
bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic {name} --describe
# 修改一个 Topic 的属性：分区变更为2个（分区只能加，不能减少，对于消费者分不清数据所属）
bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic {name} --alter --partitions 2

4.2 Consumer

# 消费者订阅测试；新打开一个终端，连接到服务器，消费者订阅一个topic，接收消息测试
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {name} --from-beginning

4.3 Producer

# 生产者发送消息；新开一个终端窗体，连接到服务器，用来生产者发送消息（回车后输入要发送的消息内容）
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic {name}

切换到 消费者终端窗口，查看接收到订阅的 topic 消息。

五、集群部署

5.1 集群配置

这里用三台机组成集群模式

• 把解压后的 /usr/local/kafka_2.13-3.2.3 文件夹分发到集群中的每个服务器
• 配置文件 config/server.properties | config/kraft/server.properties 的配置
• 启动每台服务器的 kafka

安装包分发到每台服务器

# 把解压后的 kafka 根目录文件夹分发到集群中的每台服务器
scp -r /usr/local/kafka_2.13-3.2.3 {user}@{服务器IP} /usr/local

5.2 zookeeper 方式

以 config/server.properties 文件为主的配置及启动

zookeeper 使用的端口：2181；kafka 使用的端口：9092。

# 新终端窗口登录到每台服务器，配置每台服务器的 config/server.properties
#
#
# 切换到每台服务器的 kafka 根目录
cd /usr/local/kafka_2.13-3.2.3
#
# 编辑每台配置文件 config/server.properties
vi config/server.properties
# 以下两点必须的变更
# 	1、确保每台 config/server.properties 中的 broker.id 唯一（如：0，1，2）
#	2、确保每台 config/server.properties 中的 zookeeper.connect 都相同
#	示例：zookeeper.connect={serverA}:2181,{serverB}:2181,{serverB}:2181
#
#
# 启动每台服务器的 zookeeper，kafka
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
bin/kafka-server-start.sh -daemon config/server.properties
#
# 按顺序 停止 kafka 及 zookeeper
bin/kafka-server-stop.sh && bin/zookeeper-server-stop.sh

5.3 kraft 方式

以 config/kraft/server.properties 文件为主的配置及启动

kraft 使用的端口：9093；kafka 使用的端口：9092。

# 新终端窗口登录到每台服务器，配置每台服务器的 config/kraft/server.properties
#
#
# 切换到每台服务器的 kafka 根目录
cd /usr/local/kafka_2.13-3.2.3
#
# 编辑每台配置文件 config/kraft/server.properties
vi config/kraft/server.properties
#以下三点必须的变更
#
# 	1、确保每台 config/kraft/server.properties 中的 node.id 唯一（如：1，2，3）
#	示例：node.id=1
#
#	2、确保每台 config/kraft/server.properties 中的 listeners 配置各自的主机名称或IP
#	格式：listeners=PLAINTEXT://{localhost}:9092,CONTROLLER://{localhost}:9093
#
#	3、确保每台  config/kraft/server.properties 中的 controller.quorum.voters 都相同
#	格式：controller.quorum.voters={node.id}@{host}:{port}
#	示例：controller.quorum.voters=1@{hostnameA}:9093,2@{hostnameB}:9093,3@{hostnameC}:9093
#
#
# 启动 kafka
# 首次启动时，先为集群生成一个uuid
bin/kafka-storage.sh random-uuid
# 首次启动时，分别为每台服务器配置相同的uuid
bin/kafka-storage.sh format -t {uuid} -c config/kraft/server.properties
# 后续的启动
bin/kafka-server-start.sh -daemon config/kraft/server.properties
#
# 停止 kafka
bin/kafka-server-stop.sh