企业官网建设流程全解析

1. 为什么“批量改分区副本数”是个高频刚需

Kafka 集群跑久了，几乎每个运维或平台开发都会撞上同一类需求：一批 topic 的分区数不够用了，或者副本数配置当初拍脑袋定成了 1，现在要补到 3。单机kafka-topics.sh --alter一条条敲当然能解决，但当你面对的是几十上百个 topic、跨多个集群、还要保证业务低峰期执行时，手工操作基本等于给自己埋雷。我见过最典型的事故是：某同学用脚本循环改分区，结果把带 key 的消息顺序打乱了，下游消费逻辑直接错乱，排查了大半天才发现是分区扩容导致 key 的哈希落点变了。

这篇内容就是围绕“批量修改 Kafka topic 分区副本数”这件事展开的。它本质上是一个运维自动化脚本的工程实践问题，涉及 Kafka 的 Admin API、分区扩容的语义边界、副本重分配的底层机制，以及批量操作时的幂等与回滚设计。适合谁看？如果你是会写点 Java 或 Python、平时负责 Kafka 集群维护、或者正在做中间件平台化工具的后端同学，这篇能直接拿去改改就用；如果你只是偶尔改一两个 topic，也能从里面搞清楚哪些操作是安全的、哪些是会踩坑的。

先把核心结论摆前面：分区数只能增不能减，副本数可以增减但要走重分配流程，批量操作必须做前置校验和分批限流。这三句话是整个方案的地基，后面所有代码和流程都是围绕它们展开的。我下面会从设计思路、核心机制、代码实现、问题排查四个层面拆开讲，尽量把每个“为什么这么写”都交代清楚，而不是甩一段代码让你自己猜。

2. 整体方案设计与技术选型拆解

2.1 需求边界先划清楚：改分区和改副本是两件事

很多人把“改分区副本数”当成一个动作，实际上在 Kafka 的语义里这是两条完全不同的路径。分区数变更走的是createPartitions，本质是给 topic 增加新的分区元数据，已有分区的数据不动；副本数变更走的是alterPartitionReassignments（新版本）或kafka-reassign-partitions.sh（命令行），本质是重新计算每个分区的副本分布并触发数据搬迁。这两件事的代价差了一个数量级：扩分区几乎是元数据操作，秒级完成；改副本涉及跨 broker 的数据复制，TB 级 topic 可能要跑几个小时。

所以方案设计的第一步是把输入参数拆成两组：targetPartitions和targetReplicationFactor，允许只改其中一个。如果两个都填，执行顺序必须是先扩分区、再改副本。原因很实在：扩分区会产生新的分区，这些新分区默认继承当前副本数，如果你先改副本再扩分区，新分区又得再走一遍重分配，白白多搬一次数据。这个顺序细节在官方文档里不会明说，但实际跑一遍就能体会到差别。

2.2 为什么选 AdminClient 而不是 shell 脚本

命令行工具kafka-topics.sh和kafka-reassign-partitions.sh当然能用，但批量场景下它们有几个硬伤。第一是解析成本高，你得靠 grep/awk 从文本输出里抠结果，topic 名里带特殊字符就容易翻车；第二是错误处理弱，脚本循环里某一条失败了你很难拿到结构化的异常信息；第三是重分配需要先生成 JSON 计划、再执行、再验证，三步之间状态容易丢。

用 Kafka 自带的AdminClient（Java）或者kafka-python/confluent-kafka（Python）就干净很多。以 Java AdminClient 为例，createPartitions和alterPartitionReassignments都是原生方法，返回KafkaFuture，可以精确控制超时、拿到每个 topic 的错误码。我个人的选型习惯是：平台化工具用 Java AdminClient，临时批处理用 Python 脚本。前者适合集成进服务，后者适合运维同学随手跑。下面代码示例我以 Java 为主，因为它的 API 语义最完整，Python 版本逻辑完全对应，改起来不难。

2.3 批量操作的三个设计原则

在动手写代码之前，有三个原则必须先定下来，否则脚本越写越乱。

原则一：幂等优先。脚本要能重复执行而不产生副作用。具体做法是每次操作前先describeTopics拿到当前分区数和副本数，和目标值比对，已经达标就跳过。这样即使上次执行到一半中断，重跑也不会重复扩容。

原则二：分批限流。一次性对上百个 topic 发起重分配，broker 的复制线程池会被打满，直接影响正常读写。我的经验值是每批 5 到 10 个 topic，批间 sleep 几秒，具体看集群规模。副本重分配还可以通过--throttle参数限速，AdminClient 里对应设置replicaAssignmentThrottle。

原则三：可观测与可回滚。每个 topic 的操作结果要落日志，记录操作前状态、目标状态、实际结果、耗时。改副本前最好把原始的分区副本分配方案 dump 一份出来，万一新分布导致热点，能照着原方案改回去。

3. 核心机制解析：分区与副本到底怎么变

3.1 分区扩容的语义与 key 顺序问题

createPartitions的行为很直接：你告诉它 topic 要变成 N 个分区，它在原有分区基础上追加新分区，编号从旧的最大值往后排。已有分区的数据一条都不会动，新分区一开始是空的，生产者后续写入时按 key 哈希或轮询落到新分区上。

这里有个必须讲清楚的点：如果你的生产者是按 key 分区且下游依赖同一 key 的顺序，扩分区会打破这个保证。因为 key 到分区的映射是hash(key) % 分区数，分区数一变，同一个 key 的落点就变了。扩分区之前的老数据还在老分区，新数据去了新分区，消费端如果按分区顺序处理，就会看到同一个 key 的消息乱序。这不是 bug，是分区模型的固有特性。所以脚本里我加了一个开关allowPartitionIncrease，默认 false，需要人工确认业务不依赖 key 顺序才打开。

另外，Kafka 对分区数有上限约束，单 topic 分区数受 broker 数量和num.partitions配置影响，扩之前最好确认目标值不超过集群承载能力。经验上单 broker 承载的分区总数控制在 2000 到 4000 比较稳，超过之后 controller 的元数据同步会变慢。

3.2 副本重分配的底层流程

副本数变更比扩分区复杂得多。当你把副本数从 1 改成 3，Kafka 需要为每个分区选出新的副本 broker，把现有数据复制过去，等新副本追上 leader 的 LEO 后再切换 ISR。整个过程分三步：生成重分配方案、执行重分配、验证完成。

生成方案时，Kafka 的默认策略是尽量把副本分散到不同 broker 上，同时兼顾机架感知（如果配了broker.rack）。但默认策略不保证均衡，尤其是集群 broker 数量少、topic 多的时候，容易出现某些 broker 副本扎堆。所以生产环境我建议自己算分配方案，用轮询或按负载加权的方式指定每个分区的 replica 列表，而不是完全交给默认算法。

执行阶段，数据搬迁会占用 broker 间的网络带宽和磁盘 IO。这就是为什么要限流。replicaAssignmentThrottle的单位是字节每秒，设置多少合适？一个粗略的算法是：单 broker 可用带宽的 30% 除以并发搬迁的分区数。比如 broker 间是万兆网，可用 10Gbps，留 30% 即 3Gbps 给搬迁，同时搬 10 个分区，每个分区限到 300Mbps 左右比较安全。实际值还要看磁盘 IO 能不能跟上，机械盘和 SSD 差别很大。

3.3 副本数减少的风险点

副本数从 3 减到 2 或 1 在 API 层面是允许的，但风险明显更高。减少副本意味着要下线某些 broker 上的副本，如果此时正好有 broker 宕机，可能直接导致分区不可用。所以脚本对减少副本的操作要额外加一道检查：确认目标副本数小于当前值时，先校验集群当前是否有 broker 处于离线状态，有就拒绝执行。这个检查用describeCluster拿节点列表，对比describeTopics里每个分区的 ISR 是否完整就能判断。

4. 代码实现：从单 topic 操作到批量编排

4.1 环境与依赖准备

Java 方案依赖kafka-clients，版本建议和集群版本对齐，至少 2.4 以上才支持alterPartitionReassignments。Maven 依赖如下：

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.6.0</version> </dependency>

配置文件我习惯用一个 properties 文件管理，避免硬编码：

bootstrap.servers=broker1:9092,broker2:9092,broker3:9092 batch.size=8 batch.interval.ms=5000 reassign.throttle.bytes.per.sec=314572800 operation.timeout.ms=600000

batch.size控制每批处理多少个 topic，reassign.throttle.bytes.per.sec就是前面算的限流值，300MB 左右。operation.timeout.ms给单批操作留足时间，重分配是异步的，这个超时主要针对元数据操作。

4.2 核心工具类：查询、扩容、重分配

先写一个查询当前状态的方法，这是所有幂等判断的基础：

public Map<String, TopicState> describeTopics(AdminClient admin, List<String> topics) throws ExecutionException, InterruptedException { DescribeTopicsResult result = admin.describeTopics(topics); Map<String, TopicState> states = new HashMap<>(); for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : result.topicNameValues().entrySet()) { TopicDescription desc = entry.getValue().get(); int partitions = desc.partitions().size(); int replication = desc.partitions().get(0).replicas().size(); states.put(entry.getKey(), new TopicState(partitions, replication)); } return states; }

注意这里取副本数用的是第一个分区的 replica 数量，前提是 topic 内所有分区副本数一致。如果历史上有过不均匀的重分配，各分区副本数可能不同，稳妥做法是遍历所有分区取最小值或分别记录。我在实际项目里遇到过副本数不一致的 topic，就是早期手工操作留下的，所以查询方法里加了个校验：如果各分区副本数不一致，打警告日志并跳过该 topic，让人工介入。

扩容方法很直接：

public void increasePartitions(AdminClient admin, String topic, int target) throws ExecutionException, InterruptedException { Map<String, NewPartitions> request = Collections.singletonMap( topic, NewPartitions.increaseTo(target)); admin.createPartitions(request).all().get(); }

重分配稍微绕一点，需要构造NewPartitionReassignment，指定每个分区新的 replica 列表：

public void reassignReplicas(AdminClient admin, String topic, int targetReplication, List<Integer> brokerIds) throws ExecutionException, InterruptedException { TopicDescription desc = admin.describeTopics( Collections.singletonList(topic)).allTopicNames().get().get(topic); Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>(); int idx = 0; for (TopicPartitionInfo p : desc.partitions()) { List<Integer> newReplicas = new ArrayList<>(); for (int i = 0; i < targetReplication; i++) { newReplicas.add(brokerIds.get((idx + i) % brokerIds.size())); } reassignments.put(new TopicPartition(topic, p.partition()), Optional.of(new NewPartitionReassignment(newReplicas))); idx++; } admin.alterPartitionReassignments(reassignments).all().get(); }

这段代码用的是轮询分配，idx随分区递增，保证相邻分区的 leader 落在不同 broker 上，避免热点集中。broker 列表从describeCluster动态获取，不要写死，集群扩缩容后脚本还能用。

4.3 批量编排与分批限流

把上面的方法串起来，主流程就是一个带批处理的循环：

public void batchModify(List<String> topics, int targetPartitions, int targetReplication) throws Exception { Properties props = loadProps(); try (AdminClient admin = AdminClient.create(props)) { List<Integer> brokerIds = admin.describeCluster().nodes().get() .stream().map(Node::id).collect(Collectors.toList()); int batchSize = Integer.parseInt(props.getProperty("batch.size")); long interval = Long.parseLong(props.getProperty("batch.interval.ms")); for (int i = 0; i < topics.size(); i += batchSize) { List<String> batch = topics.subList(i, Math.min(i + batchSize, topics.size())); Map<String, TopicState> current = describeTopics(admin, batch); for (String topic : batch) { TopicState state = current.get(topic); try { if (targetPartitions > state.partitions) { increasePartitions(admin, topic, targetPartitions); } if (targetReplication != state.replication) { reassignReplicas(admin, topic, targetReplication, brokerIds); } log.info("topic={} done, before={}, after target={}/{}", topic, state, targetPartitions, targetReplication); } catch (Exception e) { log.error("topic={} failed", topic, e); } } Thread.sleep(interval); } } }

这里每个 topic 单独 try-catch，保证一个失败不影响同批其他 topic。批间 sleep 是硬性要求，别图快去掉，我见过有人把 sleep 删了跑全量，结果 broker 的kafka-log-dir磁盘 IO 直接打满，正常生产的写入延迟飙到秒级。

4.4 重分配进度跟踪

alterPartitionReassignments是异步的，调用返回不代表数据搬完了。要确认完成，得轮询listPartitionReassignments：

public boolean waitForReassignment(AdminClient admin, String topic, long timeoutMs) throws Exception { long deadline = System.currentTimeMillis() + timeoutMs; while (System.currentTimeMillis() < deadline) { ListPartitionReassignmentsResult r = admin.listPartitionReassignments(); boolean done = r.reassignments().get().isEmpty(); if (done) return true; Thread.sleep(5000); } return false; }

reassignments()返回空 map 表示所有重分配都完成了。实际使用中建议把这个轮询和日志结合，每次打印还剩多少分区在搬，方便判断进度。超时时间按数据量估，一个 100GB 的分区在限流 300Mbps 下大概要 45 分钟，别把超时设太短。

5. 实操过程与关键环节实录

5.1 一次完整的批量操作流程

假设现在有一个 20 个 topic 的列表，要把分区从 6 扩到 12，副本从 1 改成 3。我的实际操作顺序是这样的。

第一步，先跑 dry-run 模式。脚本里加一个dryRun开关，只做查询和比对，打印出每个 topic 的当前状态和将要执行的动作，不真正调用变更 API。这一步能提前发现 topic 名拼错、目标值填反之类的问题。dry-run 的输出我一般会存一份文件，作为操作前的基线快照。

第二步，确认集群健康。用describeCluster看 broker 是否全部在线，用listPartitionReassignments确认没有正在进行的重分配（有的话先等它完成，别叠加）。这一步花不了两分钟，但能避免很多莫名其妙的失败。

第三步，正式执行。先只改分区，观察几分钟，确认新分区正常创建、生产者写入分布均匀，再进入副本重分配阶段。分区扩容和副本重分配分开跑，出问题时定位范围小很多。

第四步，跟踪重分配进度。每 30 秒打一次日志，记录剩余待搬分区数。如果发现某个 topic 卡住不动超过预期时间，大概率是限流太狠或者某个 broker 磁盘满了，需要单独排查。

第五步，验证结果。全部完成后重新describeTopics，确认每个 topic 的分区数和副本数都达标，且每个分区的 ISR 列表长度等于副本数（说明所有副本都同步完成）。

5.2 参数选择的计算示例

限流值到底设多少，我给一个实际算过的例子。假设集群 6 个 broker，broker 间万兆网络，实际可用带宽按 60% 算约 6Gbps。重分配时数据会在 broker 间流动，假设同时有 3 个 broker 参与搬迁，每个 broker 分到 2Gbps。再留一半余量给正常业务，每个 broker 用于搬迁的带宽约 1Gbps，即 125MB/s。如果一批同时搬 8 个分区，每个分区限流设 15MB/s 左右比较稳妥。

换算成配置就是reassign.throttle.bytes.per.sec=15728640。这个值偏保守，但生产环境宁可慢一点也别影响业务。如果集群是空闲的、或者搬迁窗口在凌晨，可以适当放大到 50MB/s 每分区。

分区数的目标值怎么定？一个参考公式是：目标分区数 = 峰值吞吐 / 单分区吞吐上限。单分区吞吐上限经验值写入 10MB/s、读取 20MB/s 左右（受页缓存和磁盘影响）。比如某 topic 峰值写入 100MB/s，那至少 10 个分区。再考虑未来半年增长，留 50% 余量，定 15 个分区比较合理。别一上来就定几百个分区，分区多了 controller 元数据压力大，而且小文件多影响顺序写性能。

5.3 操作现场的一个真实教训

有次给一个日志类 topic 扩分区，从 12 扩到 24，脚本跑完看着一切正常。结果第二天业务反馈消费延迟暴涨。排查发现这个 topic 的消费者组用的是 RangeAssignor，扩分区后重新分配，部分消费者分到的分区数翻倍，而消费线程池没扩容，单线程处理不过来。问题不在 Kafka 侧，而在消费端没做好扩容预案。

这个教训让我在脚本里加了一条：扩分区前检查该 topic 关联的消费者组，把组 ID 和当前成员数打到日志里。虽然脚本没法自动扩消费者的线程，但至少能提醒操作者去确认消费端能不能扛住。查询消费者组用admin.listConsumerGroups()和describeConsumerGroups()，把信息附在 dry-run 报告里，成本很低但价值很高。

6. 常见问题与排查技巧实录

6.1 典型报错与处理对照表

批量操作中最常撞上的错误就那么几类，我把它们整理成一张速查表，遇到时直接对号入座。

这张表里最容易被忽略的是ReassignmentInProgressException。批量脚本如果没做前置检查，对同一个 topic 重复触发重分配就会报这个。解决办法就是前面说的幂等设计：操作前先listPartitionReassignments，有进行中的就跳过并记日志。

6.2 重分配卡住不动的排查思路

重分配进度长时间不推进，按这个顺序查。先看listPartitionReassignments返回的每个分区的addingReplicas和removingReplicas，确认卡在哪个分区、哪个 broker。然后登录对应 broker 看日志，搜Shrinking ISR或Partition reassignment相关记录。常见原因有三个：目标 broker 磁盘使用率超过阈值（Kafka 默认 85% 会停止接收新副本）、broker 间网络抖动导致复制反复超时、限流值设得太低导致进度肉眼不可见。

磁盘满这个坑特别隐蔽，因为报错不会直接说“磁盘满”，而是副本复制一直重试。我的习惯是在脚本里加一个前置检查：通过 JMX 或kafka-log-dirs.sh拿各 broker 的磁盘使用率，超过 80% 就拒绝执行重分配并告警。这个检查放在 dry-run 阶段做，比事后排查省事得多。

6.3 几个只有踩过才知道的细节

细节一：扩分区后新分区的副本分布可能不均。createPartitions给新分区分配副本时用的是默认策略，如果集群 broker 负载本身不均，新分区可能全落在某几个 broker 上。解决办法是扩完分区后，对新分区单独做一次重分配，用轮询方式重新指定 replica 列表。这个操作在脚本里可以作为可选步骤。

细节二：重分配期间不要做 broker 滚动重启。数据搬迁过程中重启 broker，会导致重分配中断，恢复后可能要从头再来。运维窗口要错开，先完成重分配再重启。

细节三：alterPartitionReassignments传空 Optional 是取消重分配。这个 API 设计有点反直觉，Optional.empty()表示取消该分区的重分配，Optional.of(...)才是设置新分配。写代码时别把取消和设置搞混，我见过有人误传空值把正在跑的重分配全取消了。

细节四：大批量操作要留操作审计。每个 topic 的操作前后状态、操作人、时间戳都落一份结构化日志（JSON 格式），出问题时能快速回溯。这份日志比 Kafka 自身的__consumer_offsets之类有用得多，因为它是从操作视角记录的。

7. 脚本工程化的一些延伸想法

把上面这些拼起来，其实已经是一个能用的批量工具了。但如果要长期维护，还有几个方向值得投入。一是把 topic 列表的来源做成可配置，支持从文件读、从正则匹配、从 CMDB 接口拉，避免每次手工维护列表。二是把 dry-run 报告做成 HTML 或 Markdown 输出，附上每个 topic 的变更前后对比和风险提示，方便评审。三是接入告警，重分配超时或失败时自动通知，而不是等人去翻日志。

我个人在实际操作中的体会是，这类运维脚本的价值不在于代码多精巧，而在于把边界条件和失败路径想全。分区只能增不能减、副本数不能超过 broker 数、重分配要限流要跟踪、扩分区可能影响 key 顺序，这些约束每一条都是踩过坑之后才刻进肌肉记忆的。代码写出来可能就两三百行，但背后对 Kafka 分区模型和副本机制的理解，才是让它跑得稳的关键。最后再分享一个小技巧：任何批量变更前，先把目标 topic 的配置用describeConfigs导一份出来存档，改完对比一下，能发现很多意料之外的配置漂移。