Henry's xyz

外观

Kafka Topic 管理 ​

创建 Topic ​

shell
# 基本创建
kafka-topics.sh --create \
  --topic orders \
  --partitions 3 \
  --replication-factor 3 \
  --bootstrap-server localhost:9092

# 带配置项
kafka-topics.sh --create \
  --topic orders \
  --partitions 6 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config max.message.bytes=1048576 \
  --config cleanup.policy=delete \
  --bootstrap-server localhost:9092
参数说明
--partitions分区数，决定并行度和吞吐上限
--replication-factor副本数，≤ Broker 数
--config retention.ms消息保留时间（ms），默认 7 天
--config retention.bytes分区级别最大存储字节数
--config max.message.bytes单条消息最大字节数
--config cleanup.policydelete（按时间/大小删除）或 compact（按 Key 压缩）

查看 Topic ​

shell
# 列出所有 Topic
kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看详情
kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092

# 查看特定 Topic 配置
kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092 \
  --topics-with-overrides
shell
# describe 输出示例
# Topic: orders  PartitionCount: 6  ReplicationFactor: 3
#   Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
#   Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1
字段含义
Leader当前负责读写的 Broker ID
Replicas所有副本所在的 Broker ID 列表
Isr已同步的副本列表（In-Sync Replicas）

修改 Topic ​

shell
# 增加分区（只能增不能减）
kafka-topics.sh --alter --topic orders \
  --partitions 12 --bootstrap-server localhost:9092

# 修改配置
kafka-configs.sh --alter --topic orders \
  --add-config retention.ms=86400000,max.message.bytes=2097152 \
  --bootstrap-server localhost:9092

# 删除配置（恢复默认）
kafka-configs.sh --alter --topic orders \
  --delete-config retention.ms --bootstrap-server localhost:9092

# 查看 Topic 配置
kafka-configs.sh --describe --topic orders --bootstrap-server localhost:9092

分区数确定后应充分考虑未来扩容需求。减少分区不支持（需删除重建）。

删除 Topic ​

shell
kafka-topics.sh --delete --topic orders --bootstrap-server localhost:9092

# 批量删除（正则匹配）
kafka-topics.sh --delete --topic "test.*" --bootstrap-server localhost:9092

分区管理 ​

分区数规划 ​

因素建议
目标吞吐Partitions = max(目标吞吐 / 单分区吞吐, Consumer 实例数)
单分区吞吐一般 10-50 MB/s（视硬件和配置）
并行度≥ Consumer Group 中 Consumer 最大数量
上限避免过多分区，增加 Rebalance 和元数据开销

分区再分配 ​

shell
# 1. 生成迁移计划 JSON
cat > reassign.json <<EOF
{
  "version": 1,
  "partitions": [
    {"topic": "orders", "partition": 0, "replicas": [1, 2, 3]},
    {"topic": "orders", "partition": 1, "replicas": [2, 3, 1]}
  ]
}
EOF

# 2. 执行再分配
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json --execute

# 3. 查看进度
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json --verify

分区 Leader 再均衡 ​

shell
# 自动均衡所有 Topic 的 Leader
kafka-leader-election.sh --bootstrap-server localhost:9092 \
  --election-type preferred --topic orders --partition 0

消息管理 ​

查看消息 ​

shell
# 从指定 Offset 开始消费
kafka-console-consumer.sh --topic orders --partition 0 \
  --offset 100 --max-messages 10 --bootstrap-server localhost:9092

# 消费并查看 Key 和时间戳
kafka-console-consumer.sh --topic orders --from-beginning \
  --property print.key=true --property print.timestamp=true \
  --bootstrap-server localhost:9092

查看 Consumer Group 消费位置 ​

shell
# 列出所有 Consumer Group
kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 查看 Group 消费详情
kafka-consumer-groups.sh --describe --group order-service \
  --bootstrap-server localhost:9092

# 重置 Offset
kafka-consumer-groups.sh --group order-service \
  --topic orders --reset-offsets --to-earliest \
  --execute --bootstrap-server localhost:9092

输出含义：

字段含义
CURRENT-OFFSETConsumer 当前消费位置
LOG-END-OFFSETPartition 最新消息位置
LAG消费延迟（LOG-END - CURRENT），越小越好

常用配置 ​

配置默认值说明
retention.ms604800000（7 天）消息保留时间
retention.bytes-1（无限制）分区级别最大容量
cleanup.policydeletedelete 或 compact
delete.retention.ms86400000（1 天）Compact 后墓碑记录保留时间
min.cleanable.dirty.ratio0.5Compact 触发阈值（脏数据比例）
segment.ms604800000（7 天）日志段滚动时间
segment.bytes1073741824（1GB）日志段最大字节数
compression.typeproducer压缩类型（Broker 侧）

参考 ​