Kafka 的基本操作(1)
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
:指定消费组,用于消息分配和进度跟踪。
常见场景示例
- 测试消息流程:启动 ZooKeeper → 启动 Broker → 创建主题 → 启动生产者发送消息 → 启动消费者接收消息。
- 负载均衡:同一消费组启动多个消费者,消息会被均匀分配到不同消费者(按分区分配)。
- 广播消息:不同消费组消费同一主题,每个组都会收到全量消息。
通过以上操作,可快速上手 Kafka 的核心功能,实际开发中需结合客户端 SDK(如 Java、Python 等)实现更复杂的生产消费逻辑。