Kafka从入门到上天这个系列写到第九篇主题越来越接近硬核区域了。之前我们聊过怎么搭集群、怎么用命令行收发消息这些都是“表面功夫”。这篇我打算换个姿势直接拿创建主题这个动作开刀把背后的代码路径、分区副本分配策略以及整个底层流程翻出来聊透。你如果正在做Kafka运维调优或者准备面试被问到“create topic之后到底发生了什么”又或者只是想把Kafka原理补扎实这篇文章应该能给你一把能落地的钥匙。先说结论创建主题看似是发一条命令或调一次API实际上牵涉到客户端请求构造、Controller节点选主、元数据持久化、副本分配计算、Leader和ISR初始化等多个环节。任何一个环节没弄明白生产环境里就可能出现“主题建好了但数据就是写不进去”“分区副本全挤在一个broker上”“删除重建主题后元数据不一致”等一堆问题。1. 主题创建到底在解决什么问题1.1 主题是存储、并行和复制的核心单元Kafka里的Topic不是一张传统意义上的表而是一个逻辑上的消息分类。真正存储消息的载体是Partition一个主题可以拆成多个分区每个分区又可以有多个副本。创建主题这个动作实际上是在做一件看起来很简单、但分布式味道很重的事决定这些分区各放多少份分别放在集群里的哪些broker上以及谁来做Leader。我在刚开始接触Kafka时有一段时间没转过弯总觉得建Topic就跟建数据库表一样建完就能写。但表结构只是一个定义而Topic创建会立即触发集群内的资源分配和副本同步。尤其是分区数、副本因子、机架信息这些参数一旦写进元数据后续想改就得靠新增分区或者重新分配副本成本比改表结构大得多。1.2 它跟“建表”最核心的差别在哪里数据库建表主要是定义schema物理存储位置一般由存储引擎自己决定。但Kafka建主题时客户端需要明确告诉集群这个主题要多少分区每个分区要多少副本每个分区副本尽量怎么摊开。这就带来两个核心问题分区副本的分配是否均匀直接决定集群负载是否均衡。副本分配是否考虑机架和故障域直接决定集群的容灾能力。所以创建主题不是一个简单的DDL操作而是一次分布式资源编排。理解了这一点后面看代码和分配策略时就不会觉得绕。2. AdminClient创建主题代码简析与核心参数2.1 一段最小可运行的创建代码直接上Java代码这是生产里最常用的方式命令行底层也是这么走的。假设我们要创建名为order-events的主题12个分区3个副本保留一天消息。import org.apache.kafka.clients.admin.*; import org.apache.kafka.common.config.TopicConfig; import java.util.*; import java.util.concurrent.TimeUnit; public class CreateTopicExample { public static void main(String[] args) throws Exception { Properties props new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, node1:9092,node2:9092,node3:9092); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); try (AdminClient admin AdminClient.create(props)) { NewTopic topic new NewTopic(order-events, 12, (short) 3); MapString, String configs new HashMap(); configs.put(TopicConfig.RETENTION_MS_CONFIG, 86400000); configs.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, 2); topic.configs(configs); CreateTopicsResult result admin.createTopics(Collections.singleton(topic)); result.all().get(30, TimeUnit.SECONDS); } } }这段代码里最有信息量的就两处NewTopic(order-events, 12, (short) 3)表示分区数和副本数configs里可以带主题级配置。生产环境建议至少把min.insync.replicas设置成2否则当acksall时只要Leader副本自己写成功就会返回成功极端情况下数据丢失风险会明显放大。2.2 KafkaAdminClient在背后做了哪些事很多人以为AdminClient发请求就是简单地把建主题的请求丢给任意一个broker。实际上KafkaAdminClient内部是一个异步、带超时控制、具备元数据刷新能力的客户端。调用createTopics之后它不会直接返回一个最终状态而是马上返回一个CreateTopicsResult里面包含每个主题对应的KafkaFuture。真正的工作在线程池里异步推进。具体过程中客户端会先向bootstrap.servers里的节点发送MetadataRequest拿到集群的Controller节点信息。如果请求发到的broker不是Controllerbroker会返回NOT_CONTROLLER错误客户端会刷新元数据然后重新把CreateTopicsRequest发给正确的Controller。这个过程对上层API是透明的但如果你在用低版本客户端或自定义协议可能会遇到“有时成功有时失败”的诡异问题其实大概率就是Controller切换期间的元数据重试。2.3 创建结果解析与异步等待的坑CreateTopicsResult提供了all()和values()两种拿结果的方式。all()会在所有主题创建结束后返回统一的Futurevalues()返回每个主题单独的Future适合批量创建时只想等其中一个的场景。实际踩坑点是很多人直接用result.all().get()不设超时。一旦某个broker响应慢或网络抖动这条线程可能挂很久。我习惯统一加30秒超时并捕获ExecutionException去解析Kafka的异常结构。try { result.all().get(30, TimeUnit.SECONDS); System.out.println(topic created); } catch (ExecutionException e) { KeeperException.Code code KeeperException.Code.get(e.getMessage()); // 或者直接打印原因判断是 AlreadyExistsException 还是其他 Throwable cause e.getCause(); if (cause instanceof org.apache.kafka.common.errors.TopicExistsException) { System.out.println(topic already exists); } else { System.out.println(create failed: cause.getMessage()); } }创建主题最常见的异常有三个TopicExistsException主题已存在InvalidReplicationFactorException副本因子大于broker数量PolicyViolationException主题名不合法或者触发了自定义创建策略。3. 分区副本分配策略默认算法、机架感知与手动指定3.1 默认分配算法尽量打散再顺手把Leader轮询开副本分配这件事目标其实很朴素让每个broker上的分区总副本数尽量接近让每个分区Leader尽量均匀分布避免一个broker既是太多分区的Leader又是副本存储热点。在ZooKeeper模式时代Kafka用的核心方法是AdminUtils.assignReplicasToBrokers。大致的思路是将所有broker排成一个数组然后从某个随机起点开始按顺序给每个分区分配第一个副本也就是Leader分配完的索引固定偏移一位再分配第二个副本。加一个replicaShift变量来错开位置避免分区之间形成重复的固定搭配。翻译成人话就是第一个分区更像从broker0开始往后排第二个分区从broker1开始往后排副本顺序也跟着错开最终在宏观上达到一种“轮询散开”的效果。到了KRaft模式时代代码实现换成了ReplicaDistributor和DefaultReplicaDistributor但核心目标并没有变。分区数除以broker数得到每个broker的平均副本数然后尽量把多出来的部分平摊到前几个broker上。你不需要背那个复杂的for循环重点是要理解两个结论默认分配是确定性的相同输入在相同集群状态下的分配结果一致默认分配只保证相对均匀不保证绝对最优。3.2 机架感知是如何参与分配的如果broker配置了broker.rack属性创建主题时Kafka会尝试把同一个分区的多个副本放到不同的机架rack上。这里的机架可以理解为机房、机柜甚至是云厂商的可用区。开启机架感知后分配算法会多一层约束不同副本尽量落在不同rack里防止一个机架断电导致所有副本同时挂掉。注意这有个前提候选broker所在的rack数量必须足够。如果副本因子是3但broker只分布在2个rack里Kafka无法做到每个副本都在不同rack它只会努力把分配结果错开做不到真正意义上的跨故障域容灾。所以生产环境设计机架时最好让broker的rack数量大于最大副本因子。机架感知的效果可以用kafka-topics.sh --describe里的输出看到。分配结果里的Rack列如果比较分散说明RackAware生效。这也是面试官喜欢问的一个点Kafka怎么保证副本不落在同一个机架。3.3 手动指定副本分配什么场景会用到绝大多数情况不需要手动指定但遇到跨机房部署、冷热集群分层、容量异构时默认策略可能不够用。Kafka提供了--replica-assignment参数允许你逐分区指定副本所在broker。kafka-topics.sh --bootstrap-server node1:9092,node2:9092 \ --create --topic special-topic \ --replica-assignment 0:1,1:2这个格式的含义是每个逗号分隔一项对应一个分区。第一项0:1表示分区0的副本放在broker 0和broker 1上第二项1:2表示分区1的副本放在broker 1和broker 2上。如果只想指定部分分区剩下的还是要靠自动分配不行这个参数一旦使用必须把所有分区都写全。手动分配最需要注意的就是副本均匀性。你手动指定了副本Kafka不会帮你再做均衡校验顶多检查broker是否存活、replicas里有没有重复。我见过有人手动分配时把10个分区的Leader全部落在broker 0上结果broker 0的带宽被打满其他broker闲置。手动分配前先画一张分区副本分布表确认Leader不扎堆。3.4 分配结果能以后改吗能但不要指望通过修改主题配置完成。Kafka提供了kafka-reassign-partitions.sh工具可以把现有分区副本迁移到新的broker组合上。它是先做好规划然后通过副本复制的方式逐步迁移期间主题还能继续读写。实际操作时建议先用--generate生成候选方案再用--execute执行最后用--verify确认完成。这里有个容易忽视的点自动生成方案只是基于当前集群状态做的合理估计不代表性能最优。如果你的broker磁盘容量差异较大应该手动调整生成方案里的配置把新主题的副本尽量放到容量大、负载低的机器上。4. 底层创建流程从客户端到Controller再到元数据4.1 完整链路一次创建主题到底发生了什么我习惯把创建流程拆成七个步骤来看客户端通过AdminClient发送CreateTopicsRequest携带主题名、分区数、副本数、配置项。请求先落在任意broker上broker发现不是Controller则返回NOT_CONTROLLER。客户端刷新元数据找到当前Controller把请求发给Controller。Controller校验主题名、副本因子、是否已有同名主题、自定义策略是否通过。Controller调用副本分配逻辑计算每个分区的副本列表。元数据写入ZooKeeper或KRaft元数据日志并触发TopicChange事件。Controller向涉及到的broker发送LeaderAndIsrRequestbroker创建分区目录和副本日志整个主题进入可用状态。这七步里最耗时的是第1步到第3步的元数据传播。如果你用一个刚启动的客户端创建主题可能发现第一次请求会报LeaderNotAvailable或Metadata refresh相关异常这就是客户端元数据尚未更新。4.2 Controller的角色为什么创建主题必须经过它Controller是Kafka集群的“大脑”负责分区Leader选举、副本状态机切换、主题增加和删除等运维操作。创建主题不是一个普通broker能私自决定的事情必须集中在Controller上做一致性决策。在ZooKeeper模式时代Controller通过ZK选主产生创建主题时要往/brokers/topics路径下写数据并有ControllerChannel向所有broker广播更新。在KRaft模式下Kafka自己用Raft协议管理元数据Controller是一个Quorum中的一个节点创建主题会被写成一条元数据日志记录然后在元数据镜像中生效。理清Controller后很多面试题都能串起来为什么Controller挂了会影响创建和删除主题因为创建主题的入口在Controller上虽然消息收发不直接经过Controller但元数据变更必须等新Controller上任后重新加载元数据。4.3 客户端什么时候能看到新主题这是刚起步时最容易困惑的点。主题创建成功后马上用同一个客户端去producer.send有可能会报“disconnected”或者重试。原因是客户端元数据还未刷新它可能只认识启动时的几个topic分区不认识新分区。Kafka客户端默认的metadata.max.age.ms是300000毫秒也就是5分钟。正常情况下因为生产或消费过程中会持续发送MetadataRequest新主题几秒内就会被感知到。但如果你用的是一个“空转”的客户端什么也不收发那等5分钟确实有可能。排查这类问题时优先用kafka-topics.sh --describe确认主题状态再查客户端日志里metadata更新日志。5. 副本创建之后leader、ISR与可靠性调优5.1 Controller下发LeaderAndIsr后发生了什么主题元数据创建完成不代表分区立即可用Controller还要向每个分区的副本broker发送LeaderAndIsrRequest把第一个副本指定为Leader把其他副本放进ISR集合。收到请求的broker会创建分区对应的本地日志目录并且等待后续数据同步。这里有个很重要的细节副本刚创建时follower的LEO和HW都是0需要从Leader拉取数据。如果主题刚创建后立刻有大量消息写入follower会从Leader那里做一次日志同步追上Leader的HW。这个过程不是瞬间完成的但通常很快。如果你看到分区在创建初期处于UnderReplicatedPartitions状态不用太紧张先确认是不是持续超过replica.lag.time.max.ms。5.2 ISR的收缩与恢复Kafka通过ISR集合来保证消息的“已提交”级别。Leader会持续跟踪follower的同步情况只要follower在replica.lag.time.max.ms内没有跟上Leader的最新偏移就会被踢出ISR。默认值30秒。在主题刚创建的场景里ISR频繁变化通常不是分区初始同步引起的而是broker负载异常、磁盘IO饱和、网络抖动等因素。我个人排查ISR异常时会依次检查GC日志、网络连接数、磁盘使用率而不是一上来就怀疑Kafka配置。创建主题时把min.insync.replicas配好本质上就是在控制“ISR集合中有多少副本才允许写入”。这个值设置成2意味着即使Leader副本还在只要ISR里只剩Leader一个副本生产端按acksall写入就会报NotEnoughReplicasException。这是保护数据安全的手段不是Kafka在给你找麻烦。5.3 主题创建阶段就能埋下的可靠性隐患我在生产环境里见过最多的主题创建问题不是代码写错而是参数拍脑袋定。第一个是副本因子设成1。开发环境没问题生产环境一旦单机磁盘故障这个分区的所有数据直接没。建议生产副本因子至少2关键业务3。第二个是分区数设得过大。分区数代表并行度但也意味着每个broker上的文件句柄、网络连接、副本同步开销同步增加。一个分区数几十上百的主题如果TPS又低完全是资源浪费。分区数尽量根据目标吞吐 / 单分区吞吐来估算不要为了“以后扩展方便”一次性搞几百个分区。第三个是消息重复问题。生产端重试是导致消息重复的常见原因客户端发送失败后重试前一条消息实际已经成功后一条重试消息就会被重复写入。这个跟创建主题没有直接关系但如果你把主题的acks从1改成了all并且加了min.insync.replicas重试窗口会变长重复概率会升高。想要降低重复只能在消费端做幂等或者开启生产端的幂等特性enable.idempotencetrue。6. 常见报错排查与实战经验6.1 创建主题时的报错速查表报错或现象直接原因排查方向TopicExistsException主题已存在确认是否要新增分区或加--if-not-exists/ 代码里捕获异常InvalidReplicationFactorException副本因子超过broker数查看集群broker列表调小副本因子或扩容PolicyViolationException命中自定义创建策略检查broker配置里的create.topic.policy.class.nameNotControllerException请求发到了非Controller节点客户端会自动重试若反复发生检查Controller切换和网络分区TimeoutException创建请求处理超时查看元数据日志、ZooKeeper/KRaft d状态、broker负载UnderReplicatedPartitions分区副本数不足ISR查看follower是否存活磁盘/GC/网络是否异常6.2 创建完主题后应该做的三个验证创建一个主题不难难的是确认它健康。我每次创建完主题都会做三件事先看describe结果kafka-topics.sh --bootstrap-server node1:9092 --describe --topic order-events重点看Leader是否分散Replicas和Isr是否一致Rack列是否满足预期。再看能不能正常生产和消费kafka-console-producer.sh --bootstrap-server node1:9092 --topic order-events kafka-console-consumer.sh --bootstrap-server node1:9092 --topic order-events --from-beginning很多坑在命令行看起来正常但真实客户端生产消费时才会暴露比如ACL、网络策略、broker端口连通性。最后看监控指标。如果你有Prometheus或JMX采集直接看kafka_server_replicamanager_underreplicatedpartitions和kafka_controller_kafkacontroller_offlinepartitionscount这两个指标。一个为零是底线。6.3 手动迁移分区的实操片段当你想把某个主题的分区从旧broker迁到新broker最稳妥的方式还是三步走。先生成分配方案kafka-reassign-partitions.sh --bootstrap-server node1:9092 \ --topics-to-move-json-file topics-to-move.json \ --broker-list 0,1,2,3 --generatetopics-to-move.json长这样{ topics: [ {topic: order-events} ], version: 1 }然后找一个rebalance.json文件里面是生成出来的目标分配方案。执行kafka-reassign-partitions.sh --bootstrap-server node1:9092 \ --reassignment-json-file reassignment.json --execute最后等执行完成后执行--verify。迁移期间主题不会被锁住但如果你有消费端在拉数据可能会看到明显的“rebalance”或延迟抖动这属于迁移副作用不是故障。6.4 关于“Kafka数据延迟高”的经验主题创建阶段就把数据延迟高的根子埋下的情况也很多。最常见的是分区数设置过少导致单个分区写入压力过大Leader处理不过来生产者acksall时确认时间变长。解决办法是把主题分区数调大但分区数不能随意增加因为Kafka不支持减少分区只能新增分区。所以“创建时选多少分区”真的要慎重。另一种常见的延迟问题是副本同步慢。如果某个follower所在broker磁盘性能差会把ISR拖成只包含Leader然后生产者请求在写入Leader成功后还要等min.insync.replicas实际耗时会显著上升。这种问题用kafka-topics.sh --describe就能看到ISR不完整。6.5 这几次实践后我留下的几个习惯创建主题前先把集群当前的broker数和rack分布列出来确认副本因子小于broker数、rack分布合理。创建主题的代码里必须设超时并且对TopicExistsException单独处理不要用一段超级宽泛的catch (Exception e)吞掉所有异常。监听主题元数据变化时使用AdminClient的describeTopics而不是裸的producer.send这样能第一时间拿到更完整的错误原因。这些习惯看起来都很小但在关键时刻能帮你少熬好几个小时。到今天这篇为止Kafka主题创建这条链路基本算闭环了。分配策略是一个“让数据均匀散开”的算法底层流程是一套围绕Controller的分布式协作而排障经验则是你把这些知识转化为生产能力的最后一公里。希望在实战里你也能把每一层的原理都变成自己的判断依据。