1、概念简介

Kafka是一个高吞吐量分布式消息系统，采用Scala和Java语言编写，它提供了快速、可扩展的、分布式、分区的和可复制的日志订阅服务。它由Producer、Broker、Consumer三部分构成。

2、核心设计

2.1、组件模块

1）Producer

消息生产者，是消息的产生的源头，负责生成消息并发送到Kafka服务器上。

2）Consumer

消息消费者，是消息的使用方，负责消费Kafka服务器上的消息。

对于消费者，不是以单独的形式存在的，每一个消费者属于一个 consumer group，一个 group 包含多个 consumer。特别需要注意的是：订阅 Topic 是以一个消费组来订阅的，发送到 Topic 的消息只会被订阅此 Topic 的每个 group 中的一个 consumer 消费。

具体说来是根据 partition 来分的，一个 Partition只能被消费组里的一个消费者消费，但是可以同时被多个消费组消费，消费组里的每个消费者是关联到一个 partition 的。因此对于一个 topic同一个 group 中不能有多于 partitions 个数的 consumer 同时消费，否则将意味着某些 consumer 将无法得到消息。

3）Broker

消息中间件处理结点，一个Kafka节点（server）就是一个broker，多个broker可以组成一个Kafka集群。

Broker不保存订阅者的状态，由订阅者自己保存。Broker没有副本机制，一旦broker宕机，该broker的消息将都不可用。

4.1）Topic

在Kafka中，消息是按Topic组织的。

在同一个 Kafka 集群中，Topic 名称必须是全局唯一的，不能重复创建相同名称的 Topic。

4.2）Offset

partition中的每个消息都有一个连续的序列号叫做offset，用于partition唯一标识一条消息。
消息存储在Kafka的Broker上，消费者拉取消息数据的过程中需要知道消息在文件中的偏移量，这个偏移量就是所谓的Offset。

5.1）Partition

topic物理上的分组，一个topic可以分为多个partition，每个partition为一个目录；多个分区可以分布到多个Broker上，也可以分布到同一台Broker上；
即一个消费者可以消费多个分区，一个分区只能给一个消费者消费；

Controller：中央控制器Control，负责管理分区和副本状态并执行管理着这些分区的重新分配。（里面涉及到partition leader 选举）
ISR：同步副本组

5.2）Segment

partition物理上由多个segment组成；
每个Partition(目录)由多个大小相等segment(段)数据文件组成，但每个段segment file消息数量不一定相等，这种特性方便old segment file快速被删除。

消费组 group

offset（偏移量）存在哪？
offset 是属于 消费组 + Topic + 分区 的。
不是属于消费者，不是属于 Topic，是属于「消费组」的！
所以：
换个组，就能从头消费
同一个组，接着上次位置消费

3、关键设计

kafka 的 topic 对应的 分区数量 一般设置为 kafka 集群 broken 的整数倍；
具体如下：

• 3 Broker 集群（最常用）
  单 Topic：3、6、9、12 分区（3 的倍数）
  集群总分区：建议 < 6000
• 6 Broker 集群
  单 Topic：6、12、18、24 分区（6 的倍数）
  集群总分区：建议 < 12000
• 1 Broker（测试 / 单机）
  分区随便设，但无高可用、无并行分布

Broker 扩容后分区怎么处理？

只加 Broker 不加分区：新 Broker 几乎分不到分区，负载依然集中在旧节点
正确做法：
先把分区数调整为 新 Broker 数的整数倍；
执行 kafka-reassign-partitions.sh 重平衡‘；

其他

1）kafka服务器的编号为0起始，number-1结束；
2）消息订阅者可以rewind back到任意位置重新进行消费，当订阅者故障时，可以选择最小的offset(id)进行重新读取消费消息。

4、操作命令

topic 相关

创建 topic

• 语法格式

kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --topic ${topic_name} \
  --partitions ${part_num} \
  --replication-factor ${copy_num} \
  --config retention.ms=604800000 \
  --config retention.bytes=1073741824 \
  --config segment.bytes=536870912 \
  --config max.message.bytes=10485760 \
  --config compression.type=snappy \
  --config cleanup.policy=delete \
  --config min.insync.replicas=2

• bytes 单位

后缀	含义	示例	等效值（字节）
b	字节	1024b	1024
k	KB（千字节）	1k	1024
m	MB（兆字节）	512m	536870912
g	GB（吉字节）	1g	1073741824
t	TB（太字节）	2t	2199023255552

• 时间单位

后缀	含义	示例	等效值
ms	毫秒	500ms	500 毫秒
s	秒	60s	60000ms
m	分钟	30m	1800000ms
h	小时	24h	86400000ms
d	天	7d	604800000ms

• 参数说明

参数	说明	示例值
--bootstrap-server	Kafka 集群地址	localhost:9092
--topic	主题名称	user-events
--partitions	分区数量	3
--replication-factor	副本因子	2
--config retention.ms	数据保留时间（毫秒）	604800000 (7天)
--config retention.bytes	数据保留大小（字节）	1073741824 (1GB)
--config segment.bytes	日志分片大小	536870912 (512MB)
--config max.message.bytes	最大消息大小	10485760 (10MB)
--config compression.type	压缩类型	snappy, gzip, lz4, zstd
--config cleanup.policy	清理策略	delete 或 compact
--config min.insync.replicas	最小同步副本数	2

• 示例写法

kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic prod-events \
  --partitions 10 \
  --replication-factor 3 \
  --config retention.ms=7d \
  --config retention.bytes=50g \
  --config segment.bytes=1g \
  --config max.message.bytes=5m \
  --config compression.type=snappy \
  --config min.insync.replicas=2

• 列出topic

kafka-topics.sh --bootstrap-server localhost:9092 --list

• 查看topic

kafka-topics.sh --bootstrap-server localhost:9092 --topic ${topic_name} --describe

查看特定前缀的 topic

kafka-topics.sh --bootstrap-server localhost:9092 --list | grep ods_ | wc -l

说明：此处的 wc -l 是统计topic 数量；

• 增加分区

kafka-topics.sh --bootstrap-server localhost:9092 \
  --topic ${topic_name} \
  --alter --partitions ${part_num}

查看 message

• 语法格式

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic ${topic_name} \
  --from-beginning \
  --max-messages 10

说明：
该指令会消费对应的 topic消息，但是因为没有加 group 参数，Kafka 会自动生成一个随机临时消费者组，看完就丢，不保存offset 偏移量；下次执行又是全新的，从头 / 从最新开始看；完全不会和业务的消费者组冲突；

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic ${topic_name} \
  --from-beginning \
  --max-messages 10 \
  --property print.offset=true \
  --property print.key=true \
  --property key.separator=" | "

数据展示如下：

offset | key | value
0      | 1   | {"before":null,"after":{"id":1,"name":"张三","age":20},"op":"c"}
1      | 2   | {"before":null,"after":{"id":2,"name":"李四","age":25},"op":"c"}
2      | 2   | {"before":{"id":2,"name":"李四","age":25},"after":{"id":2,"name":"李四","age":26},"op":"u"}
3      | 1   | {"before":{"id":1,"name":"张三","age":20},"after":null,"op":"d"}

• 参数说明

上述指令里，如果不加 --from-beginning，则消息会从最新的开始查看；

批量创建 topic

#!/bin/bash

BOOTSTRAP_SERVERS="kafka1:9092,kafka2:9092,kafka3:9092"

# 需要同步的表列表（可从 MySQL 查询导出）
TABLES=("user" "order" "payment" "product" "inventory")

# 默认分区数（根据表重要性可单独调整）
DEFAULT_PARTITIONS=6
DEFAULT_REPLICATION=3
DEFAULT_RETENTION="7d"

for table in "${TABLES[@]}"; do
    TOPIC="order_db.${table}"
    echo "Creating topic: $TOPIC"
    
    kafka-topics.sh --create \
        --bootstrap-server $BOOTSTRAP_SERVERS \
        --topic "$TOPIC" \
        --partitions $DEFAULT_PARTITIONS \
        --replication-factor $DEFAULT_REPLICATION \
        --config retention.ms=7d \
        --if-not-exists  # 幂等：存在则跳过
done

echo "All topics created successfully"

5、管理工具

可视化工具kafka-eagle
下载kafka-eagle的安装包，下载地址：https://github.com/smartloli/kafka-eagle-bin/releases

================================== over =============================================