★ FEATURED ARTICLE Kafka 集群跑久了几乎每个运维或后端开发都会撞上同一个尴尬局面当初建 topic 时随手给了 3 分区 2 副本业务量一上来消费者数量扩到 8 个结果一半在空转或者集群扩容加了 broker老 topic 的副本还是死死压在原来那几台机器上新机器闲着吃灰。手工在命令行里一个个改几十上百个 topic 改到怀疑人生还容易漏、容易错。这篇就聊聊怎么用代码把批量修改 Kafka topic 分区数和副本数这件事做扎实——不是贴一段脚本就完事而是把背后的坑、参数怎么算、异常怎么兜底都掰开讲清楚适合手里管着几十个 topic、又不想半夜被报警叫醒的同行参考。1. 先搞清楚分区和副本到底在改什么动手写代码之前得先把这两个概念在 Kafka 里的物理含义捋顺否则代码写出来也是瞎改。1.1 分区数只能加不能减的硬约束Kafka 的分区数是一个单向操作——你可以从 3 加到 12但永远没法从 12 减回 3。原因在于分区是消息有序性的基本单位消息一旦按 key 哈希落到某个分区减少分区就意味着要重新分配已有消息这会破坏 offset 的连续性也会让消费者组的位点彻底乱套。所以 Kafka 官方干脆把减分区这条路封死了。这意味着你在写批量脚本时必须先把目标分区数 当前分区数的 topic 过滤掉否则脚本跑到一半抛异常前面的改了一半后面的没改状态就乱了。我的做法是在脚本里先做一次全量扫描把每个 topic 的当前分区数、副本数拉出来和目标值做对比分成需要扩容分区分区已达标分区超标需人工确认三类只对第一类执行操作。1.2 副本数调整本质是分区重分配副本数的修改和分区数不太一样。改副本数实际上是触发一次partition reassignment分区重分配Kafka 会把指定分区的副本在 broker 之间重新摆放。这个过程是异步的你提交了重分配计划之后集群会在后台慢慢搬数据搬完之前这个 topic 处于重分配进行中的状态。这里有个关键点很多人忽略改副本数不是简单地多复制一份。当你把副本从 2 改到 3Kafka 需要选一个新的 broker 来放第三个副本同时保证同一个分区的多个副本不在同一台 broker 上否则副本就失去意义了。所以重分配计划必须由集群的现状推导出来不能拍脑袋写死。1.3 两个操作能不能一起做可以但要小心。分区扩容和副本调整可以合并成一次重分配提交Kafka 的alterPartitionReassignments接口支持同时指定新的分区列表和副本分布。不过我的经验是能分开就分开先扩分区这个操作轻量只是元数据变更等新分区被消费者正常分配、流量跑稳了再单独做副本重分配这个涉及大量数据搬迁IO 压力大。两件事叠在一起做一旦出问题很难定位是哪一步引起的。2. 为什么不用现成命令非要写代码命令行工具kafka-reassign-partitions.sh和kafka-topics.sh --alter确实能改但批量场景下它们有几个绕不过去的短板这也是我最终选择写代码的核心原因。2.1 命令行工具的批量能力太弱kafka-topics.sh --alter --partitions N一次只能改一个 topic想批量就得套 shell 循环。而副本调整更麻烦kafka-reassign-partitions.sh要求你先手写一个 JSON 格式的重分配计划文件格式长这样{ version: 1, partitions: [ {topic: order-events, partition: 0, replicas: [1, 2, 3]}, {topic: order-events, partition: 1, replicas: [2, 3, 1]} ] }几十个 topic、每个几十个分区这个 JSON 靠手写根本不现实必须用程序生成。既然都要写程序生成计划了那不如把整个流程都用代码串起来。2.2 需要事务性的预检—执行—校验闭环批量操作最怕的是改了一半。命令行工具没有预检能力你没法在真正执行前知道哪些 topic 会失败。而代码可以先做一轮 dry-run检查目标分区数是否合法、目标副本数是否超过 broker 数量、集群是否有 broker 处于下线状态。这些检查在代码里就是几个 if但能挡掉 80% 的翻车场景。2.3 需要记录和回滚依据每次重分配前代码可以把当前的副本分布快照存下来写文件或存库万一改完发现有问题至少知道原来长什么样能照着恢复。命令行工具做完就做完了啥也不留。下面这张表是我总结的命令行 vs 代码方案的对比供你判断自己的场景该走哪条路维度命令行工具代码方案单 topic 修改够用略重批量修改需套 shell 循环易错天然支持重分配计划生成手写 JSON不现实程序推导预检校验无可自定义执行快照/回滚无可实现进度跟踪需另查可集成3. 用 AdminClient 搭出批量修改的骨架Java 生态里操作 Kafka 元数据的标准入口是AdminClient它封装了 topic 管理、分区重分配、集群信息查询等一整套 API。下面按初始化—扫描—预检—执行—校验的顺序把骨架搭出来。3.1 初始化 AdminClient 与集群信息拉取第一步是把客户端建起来同时拉一份集群的 broker 列表后面推导副本分布全靠它。Properties props new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker1:9092,broker2:9092,broker3:9092); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 60000); try (AdminClient admin AdminClient.create(props)) { // 拉取集群节点信息 DescribeClusterResult cluster admin.describeCluster(); CollectionNode nodes cluster.nodes().get(30, TimeUnit.SECONDS); ListInteger brokerIds nodes.stream() .map(Node::id) .sorted() .collect(Collectors.toList()); System.out.println(在线 broker: brokerIds); }这里有个细节值得说describeCluster().nodes()返回的是当前存活的 broker。如果集群里有 broker 临时下线它不会出现在这个列表里。所以推导副本分布时一定要基于这个实时列表而不是配置文件里写死的 broker 清单否则可能把副本分配到一台已经挂掉的机器上重分配直接卡住。3.2 全量扫描 topic 的当前状态拿到 broker 列表后把需要处理的 topic 现状摸清楚。listTopics配合describeTopics能拿到每个 topic 的分区数、每个分区的副本分布。ListTopicsResult listResult admin.listTopics(new ListTopicsOptions().listInternal(false)); SetString topicNames listResult.names().get(30, TimeUnit.SECONDS); // 只处理匹配前缀的 topic避免误伤内部 topic ListString targets topicNames.stream() .filter(name - name.startsWith(biz-)) .collect(Collectors.toList()); DescribeTopicsResult describeResult admin.describeTopics(targets); MapString, TopicDescription descriptions describeResult.allTopicNames().get(60, TimeUnit.SECONDS); for (Map.EntryString, TopicDescription entry : descriptions.entrySet()) { String topic entry.getKey(); TopicDescription desc entry.getValue(); int partitionCount desc.partitions().size(); // 取第一个分区的副本数作为代表正常情况下各分区副本数一致 int replicaCount desc.partitions().get(0).replicas().size(); System.out.printf(topic%s, partitions%d, replicas%d%n, topic, partitionCount, replicaCount); }注意listInternal(false)这个参数别漏否则__consumer_offsets这类内部 topic 也会被扫进来一旦被误改消费者位点全乱后果很严重。3.3 预检把不合法的目标挡在执行之前扫描完之后进入预检环节。这一步是纯逻辑判断但价值极高。我一般会检查这几项目标分区数是否小于当前值小于则跳过并记录因为 Kafka 不支持减分区。目标副本数是否大于在线 broker 数大于则无法分配直接报错。目标副本数是否小于 1非法值拦截。topic 是否正处于重分配中通过listPartitionReassignments查询正在搬数据的 topic 先跳过避免叠加操作。// 查询正在重分配的 topic ListPartitionReassignmentsResult reassignResult admin.listPartitionReassignments(); SetString reassigningTopics reassignResult.reassignments().get(30, TimeUnit.SECONDS) .keySet().stream() .map(TopicPartitionReplica::topic) .collect(Collectors.toSet()); for (String topic : targets) { if (reassigningTopics.contains(topic)) { System.out.println(跳过重分配进行中: topic); continue; } // 其余校验逻辑... }预检结果我习惯输出成一张表人工过一眼再执行尤其是生产环境多这一步能省掉很多事后解释。4. 副本重分配计划怎么推导才靠谱这是整个批量脚本里技术含量最高的部分。分区扩容简单调个接口传个数字就行副本重分配则要自己算出每个分区的每个副本该放在哪台 broker 上。4.1 轮询分配的基本思路最朴素也最常用的策略是轮询round-robin把 broker 列表排好序对每个分区的第 i 个副本选择brokerList[(partitionId i) % brokerCount]作为落点。这样能保证同一分区的多个副本落在不同 broker 上因为 i 不同取模结果不同前提是副本数 ≤ broker 数不同分区的副本分布错开负载相对均衡。ListInteger brokers brokerIds; // 已排序 int replicaCount 3; ListTopicPartitionReplica assignments new ArrayList(); for (int partition 0; partition partitionCount; partition) { ListInteger replicas new ArrayList(); for (int i 0; i replicaCount; i) { int brokerId brokers.get((partition i) % brokers.size()); replicas.add(brokerId); } assignments.add(new TopicPartitionReplica(topic, partition, replicas)); }4.2 为什么不能无脑轮询机架感知的考量轮询在 broker 数量少、机架信息不敏感的场景下够用。但如果集群跨机架部署无脑轮询可能把同一分区的三个副本全塞进同一个机架一旦那个机架断电分区直接不可用。Kafka 本身支持机架感知broker.rack配置但重分配计划是你自己生成的Kafka 不会帮你纠正。所以更稳妥的做法是如果集群配置了机架信息先按机架分组保证每个分区的副本尽量分散到不同机架。实现上可以先按机架把 broker 分桶然后每个分区的副本从不同桶里轮流取。这块逻辑稍复杂但如果你的集群确实跨机房值得花时间做。4.3 保留原有副本的最小搬迁策略轮询有个副作用它会把所有分区的副本分布全部重排哪怕原来某些副本的位置是合理的。这会导致不必要的数据搬迁——本来只需要加一个副本结果三个副本全换了位置搬了 3 倍的数据。更聪明的做法是最小搬迁对每个分区先保留原有的副本只要它们还在线只对缺失的副本位置做补充。比如原来分区 0 的副本在 [1, 2]目标副本数是 3那就保留 1 和 2从剩下的 broker 里挑一个比如 3补上最终变成 [1, 2, 3]。这样只搬一个新副本IO 压力小得多。// 最小搬迁保留原有在线副本补充新副本 ListInteger currentReplicas desc.partitions().get(partition).replicas(); ListInteger newReplicas new ArrayList(currentReplicas); for (int brokerId : brokers) { if (newReplicas.size() replicaCount) break; if (!newReplicas.contains(brokerId)) { newReplicas.add(brokerId); } }这个策略在只加副本不换机器的场景下几乎是零搬迁实测下来对线上流量影响最小。4.4 提交重分配与进度轮询计划生成好之后通过alterPartitionReassignments提交。提交是异步的返回后需要轮询listPartitionReassignments看进度。MapTopicPartition, OptionalNewPartitionReassignment reassignments new HashMap(); for (TopicPartitionReplica r : assignments) { reassignments.put( new TopicPartition(r.topic(), r.partition()), Optional.of(new NewPartitionReassignment(r.replicas())) ); } AlterPartitionReassignmentsResult alterResult admin.alterPartitionReassignments(reassignments); alterResult.all().get(60, TimeUnit.SECONDS); // 轮询进度 while (true) { MapTopicPartition, PartitionReassignment ongoing admin.listPartitionReassignments().reassignments().get(30, TimeUnit.SECONDS); if (ongoing.isEmpty()) { System.out.println(重分配完成); break; } System.out.println(进行中分区数: ongoing.size()); Thread.sleep(5000); }轮询间隔别设太短5 到 10 秒比较合适。太频繁会给 controller 增加压力太稀疏又感知不到卡死。5. 分区扩容的独立处理与顺序编排副本重分配讲完了回头说分区扩容。它比副本调整简单但顺序编排上有讲究。5.1 扩容分区的接口调用分区扩容用createPartitions接口传入 topic 名和目标总数注意是总数不是增量。MapString, NewPartitions newPartitions new HashMap(); newPartitions.put(biz-order-events, NewPartitions.increaseTo(12)); CreatePartitionsResult result admin.createPartitions(newPartitions); result.all().get(30, TimeUnit.SECONDS);increaseTo(12)的意思是把分区数增加到 12如果当前是 8就新增 4 个。如果当前已经是 12 或更多接口会抛InvalidPartitionsException所以预检那一步的过滤是必须的。5.2 为什么先扩分区再调副本前面提过要分开做这里说清楚原因。分区扩容后新分区需要被消费者组重新分配触发 rebalance同时新分区的 leader 选举、ISR 建立都需要时间。如果这时候立刻做副本重分配两件事的元数据变更会交织在一起controller 的压力陡增而且一旦出问题你分不清是扩容引起的还是重分配引起的。我的标准流程是扩分区 → 等消费者 rebalance 完成、新分区流量正常 → 再做副本重分配。中间这个等没有固定时长看业务流量一般观察 10 到 30 分钟。5.3 分区数怎么定才不浪费分区数不是越多越好。每个分区在 broker 上都对应文件句柄和内存开销分区太多会拖慢 controller 的元数据处理。经验公式是分区数 ≈ 期望吞吐 / 单分区吞吐同时考虑消费者并行度——分区数最好是你最大消费者实例数的整数倍这样每个消费者分到的分区数均匀。举个实际例子某业务峰值 50MB/s单分区实测能扛 10MB/s那至少 5 个分区消费者最多部署 8 个实例为了让 8 个实例都能分到活分区数取 8 的倍数最终定 8 或 16。别一上来就拍个 100 分区那是给自己挖坑。6. 踩过的坑这些异常不处理会很难受代码骨架搭起来能跑但生产环境跑批量操作异常处理才是分水岭。下面几个是我实际踩过、或者见别人踩过的。6.1 重分配卡在进行中不动最常见的是重分配提交后进度一直卡着。原因通常是目标 broker 磁盘满了或者网络带宽打满数据搬不过去。代码里如果只是死等就会一直挂着。我的处理是给轮询加超时超过阈值比如 30 分钟没进展就告警并记录当前状态人工介入。还有一种情况是 broker 在重分配过程中下线Kafka 会暂停搬迁等 broker 回来。这种在日志里能看到ReplicaNotAvailableException代码里捕获后应该标记该 topic 为待重试而不是直接失败退出。6.2 副本数超过 broker 数的静默失败如果目标副本数设成 5但集群只有 3 台 brokeralterPartitionReassignments会抛InvalidReplicaAssignmentException。这个异常在批量循环里如果不捕获整个批次就中断了。所以每个 topic 的操作要包在独立的 try-catch 里单个失败不影响其他 topic最后汇总报告。6.3 内部 topic 被误伤前面提过listInternal(false)但还有一种隐蔽情况某些业务 topic 名字恰好以__开头被误判为内部 topic 跳过或者反过来。稳妥的做法是维护一个显式的白名单或前缀规则别依赖自动判断。6.4 权限不足导致的半途而废如果执行脚本的账号只有部分 topic 的 ALTER 权限跑到没权限的 topic 时会抛ClusterAuthorizationException。批量场景下建议先做一轮权限预检对每个目标 topic 尝试describeTopics能描述的基本就有改的权限具体看 ACL 配置把没权限的提前剔除。下面这张表汇总了常见异常和应对方式异常类型触发场景处理策略InvalidPartitionsException目标分区数 ≤ 当前值预检过滤跳过InvalidReplicaAssignmentException副本数 broker 数预检拦截报错ReplicaNotAvailableExceptionbroker 下线标记待重试稍后再跑ClusterAuthorizationException权限不足预检剔除记录TimeoutException重分配卡住超时告警人工介入7. 上线前的自测清单与灰度节奏批量改 Kafka 元数据最忌讳直接在生产全量跑。我一般按这个节奏来。7.1 先在测试集群跑通全流程测试集群不需要和生产一样大3 台 broker 就够验证逻辑。重点验证分区扩容后消费者能否正常 rebalance、副本重分配后 ISR 是否正常、脚本的预检和异常分支是否按预期走。测试集群上可以故意造几个异常场景比如把目标副本数设成 10看脚本是不是优雅地报错而不是崩掉。7.2 生产环境先灰度一个 topic挑一个流量小、非核心的 topic 先跑一遍观察重分配进度、集群 IO、消费者延迟。确认没问题后再按批次扩大范围每批之间留出观察窗口。我习惯每批不超过 10 个 topic批间间隔 15 分钟。7.3 执行前必备的快照每次执行前把当前所有目标 topic 的分区数、副本分布导出成 JSON 存好。万一改完发现某个 topic 的副本分布不理想至少能照着快照重新生成一份恢复计划。这个习惯救过我一次——某次轮询策略把副本全排到了两台机器上靠快照快速回滚了。# 快照示例伪代码实际用 AdminClient 导出 kafka-topics.sh --describe --topic biz-order-events snapshot_$(date %s).txt8. 几个让脚本更耐用的工程化细节最后聊几个提升脚本健壮性的点都是实际用下来觉得值的。8.1 幂等性设计脚本要能重复跑而不产生副作用。判断依据是当前状态是否已等于目标状态——分区数已达标的跳过扩容副本分布已符合目标的跳过重分配。这样即使中途失败重跑一遍也能接着往下走不用手工清理。8.2 并发度控制批量操作不要开太多并发。重分配本身就会占用大量网络和磁盘 IO如果同时对几十个 topic 提交重分配集群直接被打爆。我的做法是串行提交、异步等待一次只提交一个 topic 的重分配等它完成再提交下一个。慢是慢点但稳。8.3 日志要能定位到具体 topic 和分区批量脚本的日志如果只打操作失败排查起来很痛苦。每条日志至少带上 topic 名、分区号、操作类型、异常摘要。我一般用结构化日志方便后续 grep 或导入分析。8.4 把配置外置目标分区数、目标副本数、topic 前缀、broker 列表这些别写死在代码里抽到配置文件或命令行参数。不同环境测试、预发、生产用不同配置代码一份就够。# config.yaml 示例 target: partitionCount: 12 replicaCount: 3 filter: topicPrefix: biz- excludeTopics: - biz-temp-debug这套东西搭下来一个几十 topic 的集群批量调整从原来手工折腾大半天变成跑个脚本喝杯咖啡的功夫。真正花时间的不是写代码而是把预检、异常、快照这些防御性的部分做扎实——毕竟改 Kafka 元数据这事稳比快重要得多。我个人在实际操作中的体会是凡是涉及副本重分配的批量操作宁可多花十分钟做预检和灰度也别图快一把梭事后救火的成本远高于事前多等的那些时间。
阅读完成 · 觉得有帮助？
