Kafka-主题(Topic)介绍和使用

我们上一章介绍了中间件：Zookeeper，本章将介绍另外一个中间件：Kafka。目前这2个中间件都是基于JAVA语言。

我们上一个小节进行了Kafka的部署，本小节我们介绍Kafka里面Topic的详细介绍。

在 Apache Kafka 中，Topic（主题） 是消息的逻辑分类单元，所有消息的生产和消费都围绕 Topic 进行。每个 Topic 可以看作一个消息队列的抽象，但 Kafka 通过 分区（Partition） 和 副本（Replica） 机制实现了高吞吐、高可用和水平扩展的特性。以下是 Kafka Topic 的核心知识点：

一、Topic 的核心概念

分区（Partition）决定了 Topic 的并行处理能力（消费者组内每个消费者可消费一个分区或者多个分区，但是一个分区只能被一个组内的一个消费者消费）。分区数在创建 Topic 时指定，后续可动态增加（但不可减少）。将 Topic 的数据分割成多个有序的、可并行处理的子队列。每个分区是一个独立的日志文件（Log Segment），消息按顺序追加写入。

副本（Replica）表示每个分区的总副本数量（必须小于等于 Broker 数量）。生产环境通常设置为 3，确保数据安全性和可用性。每个分区的数据会被复制到多个 Broker 上，实现容灾和高可用。分为 Leader Replica（处理读写请求）和 Follower Replica（异步同步数据）。

偏移量（Offset）每个分区中的消息都有一个唯一的偏移量（整数），用于标识消息的位置。消费者通过维护 Offset 记录消费进度。

二、Topic 的创建与配置

创建 Topic通过命令行工具 kafka-topics.sh 创建 Topic：

./bin/kafka-topics.sh --create \
--topic my-topic \
--bootstrap-server localhost:9092 \
--partitions 3 \ 
--replication-factor 2

--partitions 分区

--replication-factor 副本数量（包含自己）

关键配置参数 在创建或修改 Topic 时，除了上面2个分区和副本核心参数外，还可指定以下参数（部分）：

• retention.ms：消息保留时间（默认 7天，可设为 -1 表示永久保留）。
• retention.bytes：消息保留的最大数据量（如 1GB）。
• cleanup.policy：数据清理策略（delete 或 compact，后者用于日志压缩）。
• compression.type：消息压缩算法（如 gzip, snappy, lz4）。
• min.insync.replicas：最小同步副本数（影响数据一致性）。

三、Topic 的存储机制

分区日志结构：每个分区在磁盘上对应一个目录，目录名为 <topic>-<partition>（如 my-topic-0）。目录中包含多个 分段日志文件（Segment），每个 Segment 包含：

.log 文件：实际存储消息。

.index 文件：消息的偏移量索引（用于快速定位消息）。

.timeindex 文件：消息的时间戳索引。

消息写入流程：生产者将消息发送到 Topic 的某个分区（通过分区策略）。Leader Replica 将消息追加到当前活跃的 Segment 文件。Follower Replica 从 Leader 异步拉取数据并保存。

消息删除与压缩：

• 删除策略（cleanup.policy=delete）：基于时间（retention.ms）或数据量（retention.bytes）删除旧消息。
• 压缩策略（cleanup.policy=compact）：保留每个 Key 的最新消息（适用于需要精确一次语义的场景，如数据库变更日志）。

四、Topic 的管理与监控

常用命令

#查看所有 Topic
./bin/kafka-topics.sh --list \
--bootstrap-server 192.168.31.143:9092

 #查看 Topic 详情   
./bin/kafka-topics.sh --describe \
 --topic my-topic \
 --bootstrap-server 192.168.31.143:9092

#增加分区数 
bin/kafka-topics.sh --alter \
--topic my-topic \
--partitions 6 \ 
--bootstrap-server 192.168.31.143:9092
#删除 Topic
bin/kafka-topics.sh --delete \
--topic my-topic \
--bootstrap-server 192.168.31.143:9092

当然我们这里的都是通过命令行来操作的，实际上一般都是通过业务api直接来创建，当然我们也可以配置不允许api直接创建，只能我们手工创建。而这个参数是auto.create.topics.enable，默认它是true。默认创建的Topic，是读取默认的分区和副本策略，开启的时候需要考虑到这个问题。