★ FEATURED ARTICLE 前阵子有个做交易系统的团队找到我提了两个问题。第一个是“Kafka单节点到底能承载多少数据量”第二个是“消费者端的负载均衡能不能根据IP来路由让指定机器消费指定数据”。这两个问题放在一起问的人不少因为大家潜意识里都希望消息中间件既能估算容量又能在消费端像负载均衡器一样做到按业务规则分发。但实际做过Kafka运维和调优的人都知道前者是一个容量规划问题后者是一个分区分配规则问题二者完全不同绝不能混为一谈。这篇文章我会把这两件事彻底拆开讲清楚。容量部分会给出存量计算的公式、吞吐压测的方法和单节点合理规划的建议路由部分会解释Kafka消费者负载均衡的真正工作机制并给出按Key路由、自定义分区器、自定义分配器、手动指定分区几种可落地方案。最后还会分享我在实际生产环境里踩过的坑包括分区扩容导致路由失效、消费组频繁Rebalance等典型问题。无论你是刚接触Kafka的新手还是已经在维护大规模集群的开发者这篇应该都能给你一些可直接参考的东西。1. 关于单节点承载量先算存量再算吞吐1.1 存量容量公式先把一个最常见的误解纠正掉Kafka单节点能承载的数据量不是一个固定的“几十万TPS”而是一个由磁盘空间、保留策略、消息大小、副本因子共同决定的容量规划问题。所谓“能存多少数据”最核心的指标是磁盘。你可以用下面这个公式先估算单日新增磁盘占用 每秒写入消息条数 × 平均单条消息大小 × 86400 × 副本因子 ÷ 压缩系数举个例子。假设业务峰值每秒写入5000条消息平均每条2KBtopic副本因子为2保留3天不开启压缩每秒写入流量 5000 × 2KB 10MB/s单日新增 10MB/s × 86400秒 ≈ 864GB副本因子2后 1.73TB/天保留3天总量 5.2TB左右如果业务消息是JSON文本且开启gzip压缩通常能压到原始体积的25%到35%那么3天窗口内的磁盘占用可以降到1.7TB上下。但注意如果消息本身是已经压缩过的图片、视频或者加密后的二进制数据压缩率基本为零必须按最坏情况估。注意这个公式只是存量估算单位是“能存多少TB”。它回答的是“磁盘够不够”回答不了“能不能扛得住这一波流量”。所以要继续看吞吐。磁盘容量不是越高越好还要考虑保留策略。很多团队习惯把retention设置得很长比如保留30天结果某天流量翻倍磁盘直接被写满整个broker开始拒绝写入。我的建议是保留窗口、副本因子、单条消息大小这三项要有专人负责定期核对任何一个变化都要重新跑一遍容量公式。1.2 吞吐到底看哪几条边吞吐的真实上限取决于瓶颈点通常是以下四个之一网络带宽例如1Gbps网卡的理论上限约125MB/s实际可用打八折大概100MB/s。如果跨机架或跨机房还要考虑专线带宽。磁盘顺序写速度Kafka的消息是先写OS PageCache再异步刷到磁盘所以磁盘的顺序写速度直接决定了极限。普通SATA SSD顺序写能到400-500MB/s机械硬盘也能到150-200MB/s但随机读写会垮掉。CPU开销消息压缩、解压、TLS加密、ACL鉴权都会消耗CPU。开启gzip压缩后CPU可能先于网络成为瓶颈。分区数与文件句柄每个分区对应一组日志段文件和若干索引文件分区数上去后broker需要管理的文件句柄暴增协调成本也会上升。网上经常能看到“Kafka单节点百万级TPS”的说法这并不假但通常是短时压测、PageCache吸收写入、无磁盘落盘压力、无副本复制的前提下测出来的。生产环境里长时间稳定运行数据还要落盘还要被消费还要容忍消费者追赶这时候能跑到的数字会低得多。分区数与吞吐不是线性增长。我在常规压测中见到的情况是单分区小消息1KB左右可以达到数万条每秒当分区数从1增加到32甚至64时整机吞吐会明显上升但如果继续加到几百个分区吞吐反而可能下降。原因在于broker线程、文件句柄、页缓存碎片等开销开始互相争抢资源。单节点上topic分区总数建议控制在两三百以内超过五百就要仔细评估了。1.3 单节点合理规划的几个数字给出几个我实际使用的参考口径不是绝对真理但可以作为初始规划依据维度参考值说明单节点稳定生产流量峰值流量的50%以下例如压测能到100MB/s生产峰值不应超过50MB/s单节点分区总数200个以内超过后注意文件句柄和Rebalance开销单分区写入速率小消息约1万-3万条/s具体依赖消息大小、批次、acks配置整机写入速率1GbE网卡下保守60-80MB/s压测99%延迟开始抖动的拐点就是警戒线磁盘占用率保持在70%以下超过80%会有磁盘写满风险Kafka可能在自动删除日志时阻塞很多人规划容量只盯消息条数其实更该盯的是MB/s。同样的10万条每秒1KB消息和10KB消息对系统的压力完全是两个量级。把线上消息的平均大小用监控统计出来再做容量估算才是正确姿势。2. 消费端的负载均衡分区才是唯一路由单位2.1 消费组和分区分配的工作原理接下来回答第二个问题“负载均衡能否根据IP路由消费”。理解这个问题的前提是先理解Kafka消费者负载均衡的单位。Kafka的消费模型里同一个消费组内的所有消费者共同订阅一个或多个topic组内协调由broker端的分区协调器负责。协调器会监控每个消费组成员的心跳当组内有成员加入、退出、订阅关系变化或者topic分区数变化时触发一次Rebalance把待消费的分区重新分配给组内的消费者。Kafka默认提供的分配策略主要有四种RangeAssignor按topic顺序排列分区按消费者顺序分段分配可能出现明显倾斜。RoundRobinAssignor轮询分配分区在消费者之间尽量均匀。StickyAssignor在Rebalance时尽量保留上一次的分配结果减少分区迁移。CooperativeStickyAssignorSticky的增强版支持增量Rebalance避免全量阻塞。最核心的认知是一个分区在同一个消费组内同一时刻只能被一个消费者消费。负载均衡的最小单位是“分区”而不是“IP”也不是“消息”。如果组内有5个消费者但topic只有3个分区那么多出来的2个消费者会空闲如果组内有2个消费者但topic有10个分区那么每个消费者会分到5个分区。Rebalance的过程大致是每个消费者启动后会向协调器发送JoinGroup请求协调器选择组内某个消费者作为leaderleader根据所有成员上报的订阅信息执行分区分配算法然后把分配结果通过协调器广播给所有成员。这里的输入信息是消费者的memberId、订阅的topic列表以及自定义元数据并不包含什么“消费者IP路由规则”。2.2 为什么“按IP路由消费”听着合理落地却坑很多很多人觉得“负载均衡根据IP路由消费”是顺理成章的事因为IP是机器最直观的标识。但在Kafka的机制里这个想法有几个硬伤。第一消费者实例并不等同于IP。Kafka消费者实例的标识是memberId由client.id加上随机UUID组成。同一台物理机上完全可能启动多个消费者进程它们IP相同但memberId不同。你把路由规则绑在IP上这多个消费者怎么区分第二IP是会变的。容器重建、DHCP重分配、虚拟机迁移都会导致消费者的IP变化。一旦IP变化按旧IP绑定的分配规则就会失效消费组会进入异常状态。第三Kafka的故障转移依赖的正是rebalance。如果分区被硬编码绑定到某个IP的消费者这台机器挂掉后其他消费者无法接管它的分区这个topic的消费就断了。真正合理的思路是分区分配只认“消费组成员”不同成员之间允许携带自定义信息分配策略可以在自定义信息上做文章。但要注意这不是简单的“把IP传给coordinator”就能解决需要自己实现分配算法。2.3 正确拆解需求按IP、按Key还是按机架在实践中“按IP路由消费”通常不是一个真实需求而是用户对需求的粗粒度描述。我一般会继续追问你到底想要哪种效果如果想让同一类业务消息比如同一用户、同一订单的所有事件都交给同一个消费者处理那么你需要的不是IP路由而是消息Key路由。如果想让某个机房或某台特定机器只消费某几个分区那么你需要的是自定义分区分配策略或手动指定分区。如果只是想减少跨机房的网络流量让消费者优先消费本机架的副本数据那么Kafka 2.4之后的client.rack配置才是更合适的方案。先把需求拆对再去选方案才不会白白写一堆代码。3. 可落地的三种路由方案3.1 方案一Producer端用Key做路由这是我认为最常用、成本最低、收益最高的方案。Kafka生产者在发送消息时可以指定一个Key默认分区器会对Key做哈希然后将消息路由到对应的分区。比如订单系统里你希望同一个用户的所有订单事件都被同一个消费者处理发送时把用户ID作为KeyProducerRecordString, String record new ProducerRecord( order-events, String.valueOf(order.getUserId()), payload ); producer.send(record);只要Key相同哈希值就相同消息就会进入同一个分区。同一个分区只会被组内同一个消费者消费所以顺序和归属都得到了保证。这个方案的好处是不改broker任何配置不写自定义分配器只需要在producer端指定Key。绝大多数业务场景包括订单、支付、风控、日志聚合都可以用这一招解决。注意默认分区器在Key不为空时使用的是Murmur2哈希不是Java的hashCode不要拿Java的hashCode去推算分区号。3.2 方案二自定义分区器如果默认的Key哈希满足不了需求比如你想让某几类大客户的消息固定落到几个特定分区方便关键消费端做隔离可以实现自定义Partitioner。public class CustomerPartitioner implements Partitioner { Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { ListPartitionInfo partitions cluster.partitionsForTopic(topic); int numPartitions partitions.size(); if (key null || numPartitions 0) { return ThreadLocalRandom.current().nextInt(Math.max(1, numPartitions)); } String businessTag (String) key; if (vip.equals(businessTag)) { return 0; } return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } Override public void close() {} Override public void configure(MapString, ? configs) {} }然后在producer端指定partitioner.classcom.example.CustomerPartitioner自定义分区器只是改变“消息进哪个分区”并没有改变消费者分配规则。想让某个消费者固定消费这个分区仍然需要在消费端配合手动指定分区或自定义分配器。注意自定义分区器一旦上线topic分区数就不能随意变更。原因很简单对所有Key取模的分区数变了老数据读出来后就找不到原来那个消费者处理了消息顺序和归属关系都会被打乱。3.3 方案三自定义ConsumerPartitionAssignor或手动assign如果确实存在“指定IP/指定机器消费指定分区”的强需求可以走两条路。第一条路是实现自定义分配策略。消费者接口提供了ConsumerPartitionAssignor可以重写subscriptionUserData()让每个消费者在加入消费组时上报自己的自定义信息比如机器ID、机架ID、IP等。分配时你可以在assign()方法里读取这些自定义信息把分区优先分配给特定成员。这里给一段简化的伪代码主要是展示思路具体接口签名要按你所用Kafka版本的官方文档来写public class RackAwareAssignor extends AbstractPartitionAssignor { Override public String name() { return rack-aware; } Override public ByteBuffer subscriptionUserData() { String rackId System.getenv(RACK_ID); return ByteBuffer.wrap(rackId.getBytes(StandardCharsets.UTF_8)); } Override public MapString, Assignment assign( MapString, Subscription subscriptions, int defaultPartitionCount, long defaultAssignmentTimeoutMs) { // 1. 遍历所有订阅的topic分区 // 2. 按subscriptionUserData()里的机架ID/实例ID对消费者分组 // 3. 将分区优先分配给同组消费者分配结果写入Assignment return assignments; } }消费者侧配置partition.assignment.strategycom.example.RackAwareAssignor这个方案能做到真正意义上的“按自定义维度做负载均衡”但实现成本不低而且要自己处理各种Rebalance边界情况。如果不是很有必要我不建议一般团队直接上。第二条路更简单粗暴用assign()手动指定分区ListTopicPartition partitions Arrays.asList( new TopicPartition(order-events, 0), new TopicPartition(order-events, 1) ); consumer.assign(partitions); consumer.seekToBeginning(partitions);这种方式完全绕开了消费组协调器不参与组内Rebalance也不会有负载均衡。它的优点是精确、可控、低延迟适合离线任务、特殊数据处理、测试环境。缺点是如果消费者挂了没有其他消费者会接管这些分区如果新增分区你需要手动扩展列表。线上业务如果追求高可用要慎用。4. 单节点容量评估实测过程4.1 压测环境与命令理论说了那么多最终还是要用数据说话。我自己做单节点容量评估时一般会准备一台和线上配置接近的机器topic设为32个分区副本因子为1关闭压缩分别处理1KB小消息和100KB大消息两种场景。先用Kafka自带的压测工具测生产者吞吐bin/kafka-producer-perf-test.sh \ --topic perf-1k \ --num-records 5000000 \ --record-size 1024 \ --throughput -1 \ --producer-props bootstrap.servers127.0.0.1:9092 acks1 linger.ms5 batch.size262144再测消费者吞吐bin/kafka-consumer-perf-test.sh \ --bootstrap-server 127.0.0.1:9092 \ --topic perf-1k \ --messages 5000000 \ --threads 1 \ --reporting-interval 1000压测的同时一定要在另一个终端盯系统指标dstat -tcmdn 1 iostat -dx 1 sar -n DEV 1重点关注四个指标网络带宽利用率、磁盘util、CPU平均负载、内存可用量。提示压测前要清理旧数据topic重新创建一次。否则PageCache里可能已经有部分数据读写混合会让结果失真。4.2 如何从压测数据推导容量比如某次压测中1KB消息、32分区、acks1生产者稳定跑到12万条每秒约117MB/s此时网卡利用率已经接近上限。继续加压延迟立刻从几十毫秒跳到几百毫秒说明网络就是瓶颈。那么生产容量建议就按峰值不超过60MB/s来规划大约6万条每秒。如果你预估业务峰值会到10万条每秒单节点就顶不住了需要扩容到两台或者开启消息压缩把体积降下来。消费者端的压测同样重要。消费者压测结果会告诉你单节点能支撑多少下游消费速率。如果LAG持续上涨要么加消费者实例要么优化消费逻辑要么扩容分区。存储容量则按1.1节公式单独计算。两个结果都出来后取更小值作为单节点的服务能力上限。记住一个原则任何硬件的理论峰值都不能作为生产承诺值至少预留50%的余量。4.3 配套参数调优清单容量评估时有些参数会影响结果这里列一下我常用的基线配置。broker端num.network.threads8 num.io.threads8 socket.send.buffer.bytes102400 socket.receive.buffer.bytes102400 socket.request.max.bytes104857600 log.retention.hours72 log.segment.bytes1073741824 unclean.leader.election.enablefalseconsumer端enable.auto.commitfalse max.poll.records500 session.timeout.ms15000 heartbeat.interval.ms3000 max.poll.interval.ms300000 partition.assignment.strategyorg.apache.kafka.clients.consumer.CooperativeStickyAssignorproducer端acks1 linger.ms5 batch.size131072 compression.typegzip要说明一下acks1能在大多数场景下保证较好的吞吐但极端故障时可能丢消息业务对可靠性要求极高应该用acksall。开启gzip压缩后如果CPU还有富余可以明显降低网络和磁盘压力。5. 踩过的坑和排查技巧5.1 分区扩容后Key路由失效有个真实教训。某业务上线时topic分了32个分区一年后业务量翻倍大家决定扩到64个分区。结果扩容后那些按用户ID做Key的消息并没有像预想中那样继续落在同一个分区而是被重新哈希到了别的分区。原因不复杂默认分区器的哈希结果对分区数取模分区数变了取模结果自然全变了。这个坑一旦踩中处理起来非常麻烦。老数据还在旧分区里新数据去了新分区消费者如果沿用旧逻辑部分历史消息会落到不匹配的处理实例上。教训是topic分区数规划时就要预留足够的增长空间比如初始64个分区之后两年内一般够用。如果确实要扩容建议新建一个topic把数据迁移过去而不是在旧topic上直接调分区数。5.2 消费组频繁Rebalance风暴当时排查一个线上问题消费组LAG突然上涨但消费者CPU并不高也没有慢查询。最后发现组内每隔一两分钟就发生一次Rebalance。原因是有个消费者处理一批消息耗时超过了max.poll.interval.ms心跳正常但poll调用超时协调器判定它已经失联把它踢出消费组触发了全量Rebalance。Rebalance期间所有消费者暂停消费LAG自然猛涨。这个问题的解法有几层把max.poll.records从默认的500调小到200甚至100控制单次poll的消息总量。消费逻辑改成异步化让poll循环快速返回不要阻塞在业务处理上。把session.timeout.ms和max.poll.interval.ms设置成合理的配合关系不要一个极小一个极大。使用CooperativeStickyAssignor减少Rebalance需要迁移的分区。5.3 如何快速查看消费组分配情况排查消费问题时最实用的命令是kafka-consumer-groups.shbin/kafka-consumer-groups.sh \ --bootstrap-server 127.0.0.1:9092 \ --describe \ --group order-group输出的核心列有CONSUMER-ID、HOST、CURRENT-OFFSET、LOG-END-OFFSET、LAG。LAG为0说明消费跟上了。LAG持续增长说明消费能力不足要么加消费者要么优化逻辑。某行CONSUMER-ID为空但LAG很大说明分区没有被任何消费者接管要么消费者数不够要么Rebalance还没完成。HOST列只是展示消费者注册时的地址Kafka分配时并不拿它当作路由依据。很多同学看到HOST列以为“IP路由”是现成的这就是误读。这个命令在排查“某分区一直没人消费”“某个消费者分到的分区太多”等问题时非常有效应该成为每个Kafka使用者的基础工具。最后再分享一点个人体会。单节点承载量没有标准答案一定要基于自己的消息大小、保留时长、副本数去压测和计算峰值按50%余量规划。路由消费这件事能走Key路由就不要写自定义分配器除非你确实有非常明确的分区绑定或机架亲和需求。Kafka的机制已经很成熟顺着机制用省心省力非要逆着机制造轮子最后往往要在故障处理上付出更大的代价。
阅读完成 · 觉得有帮助？
