当前位置：首页 > ai >正文

Kafka 的基本操作（1）

ai 2025/8/10 13:03:52

Kafka 是一个分布式流处理平台，核心功能是高吞吐量的消息发布与订阅。以下是 Kafka 最常用的基本操作，涵盖环境启动、主题管理、消息生产与消费等核心场景（基于 Kafka 2.x+ 版本，使用命令行工具）。

一、环境准备与启动

Kafka 依赖 ZooKeeper 管理集群元数据，需先启动 ZooKeeper，再启动 Kafka Broker。

1. 启动 ZooKeeper

# 进入 Kafka 安装目录
cd /path/to/kafka# 启动内置 ZooKeeper（仅用于测试，生产环境需独立部署）
bin/zookeeper-server-start.sh config/zookeeper.properties

默认端口：2181，配置文件路径：config/zookeeper.properties。

2. 启动 Kafka Broker

# 另开一个终端，启动 Kafka 服务
bin/kafka-server-start.sh config/server.properties

默认端口：9092，配置文件路径：config/server.properties（可修改端口、日志目录等）。

提示：生产环境中，可通过复制 server.properties 配置多个 Broker（修改 broker.id、port 等），组成集群。

二、主题（Topic）管理

主题是 Kafka 中消息的分类容器，所有消息必须发送到指定主题。

1. 创建主题

bin/kafka-topics.sh \--bootstrap-server localhost:9092 \  # 指定 Kafka 服务地址（新版本推荐）--create \                           # 创建操作--topic test-topic \                 # 主题名称--partitions 3 \                     # 分区数（提高并行度，建议 >=  broker 数）--replication-factor 1               # 副本数（高可用，需 <= broker 数，单机测试设为 1）

分区（Partitions）：主题的分片，消息按规则分配到不同分区，实现并行读写。
副本（Replication-factor）：每个分区的备份数量，防止单点故障（生产环境建议设为 2-3）。

2. 查看所有主题

bin/kafka-topics.sh \--bootstrap-server localhost:9092 \--list

3. 查看主题详情

bin/kafka-topics.sh \--bootstrap-server localhost:9092 \--describe \--topic test-topic

输出包含分区数量、副本分布、首领分区（Leader）等信息。

4. 修改主题（增加分区）

bin/kafka-topics.sh \--bootstrap-server localhost:9092 \--alter \                            # 修改操作--topic test-topic \--partitions 5                       # 分区数只能增加，不能减少

5. 删除主题

bin/kafka-topics.sh \--bootstrap-server localhost:9092 \--delete \--topic test-topic

注意：默认配置下，删除主题需在 server.properties 中开启 delete.topic.enable=true（默认已开启）。

三、生产消息（Producer）

通过命令行工具模拟生产者发送消息到主题。

bin/kafka-console-producer.sh \--bootstrap-server localhost:9092 \--topic test-topic \--producer-property acks=1  # 可选：消息确认机制（1 表示 Leader 接收成功即返回）

执行后进入交互模式，输入消息并回车即可发送（例如输入 hello kafka）。

四、消费消息（Consumer）

通过命令行工具模拟消费者从主题接收消息。

1. 基础消费（默认从最新消息开始）

bin/kafka-console-consumer.sh \--bootstrap-server localhost:9092 \--topic test-topic

此时若生产者发送新消息，消费者会实时接收。

2. 从头消费（包含历史消息）

bin/kafka-console-consumer.sh \--bootstrap-server localhost:9092 \--topic test-topic \--from-beginning  # 从最早的消息开始消费

3. 指定消费组（Consumer Group）

消费组是 Kafka 实现消息负载均衡和广播的核心机制，同一组内的消费者共享消息（一条消息仅被组内一个消费者消费）。

bin/kafka-console-consumer.sh \--bootstrap-server localhost:9092 \--topic test-topic \--group my-group  # 指定消费组名称

4. 查看消费组信息

# 查看所有消费组
bin/kafka-consumer-groups.sh \--bootstrap-server localhost:9092 \--list# 查看指定消费组详情（包括偏移量）
bin/kafka-consumer-groups.sh \--bootstrap-server localhost:9092 \--describe \--group my-group

输出包含每个分区的消费进度（当前偏移量、最新偏移量、滞后量）。

五、停止服务

1. 停止 Kafka Broker

bin/kafka-server-stop.sh

2. 停止 ZooKeeper

bin/zookeeper-server-stop.sh

关键参数说明

--bootstrap-server：Kafka Broker 地址（格式：host:port，多个用逗号分隔），替代旧版本的 --zookeeper（旧方式依赖 ZooKeeper，已不推荐）。
--topic：操作的主题名称。
--from-beginning：消费者从主题的起始位置开始消费。
--group：指定消费组，用于消息分配和进度跟踪。