企业官网建设流程全解析

1. 消息积压不是"磁盘满了"那么简单

很多人第一次遇到Kafka消息积压，第一反应是去看磁盘使用率，发现磁盘还有空间，就懵了。实际上消息积压的本质是消费速度跟不上生产速度，磁盘满只是最终结果，不是原因。我见过太多团队在积压发生后才开始翻监控，但那时候Consumer Lag已经飙到几百万条，业务侧早就炸了。

1.1 积压的三种典型形态

根据我这些年的排查经验，积压大致分三种情况，处理思路完全不同。

第一种是突发流量型积压。比如某电商平台做整点秒杀，平时每秒写入五千条，活动开始瞬间冲到每秒八万条。Consumer端还是按平时的节奏在拉取，Lag曲线几乎垂直上升。这种积压的特点是来得快去得也快，只要Consumer没有挂掉，活动结束后一段时间内会自然消化。但如果你的Consumer处理逻辑里有同步RPC调用或者数据库慢查询，那消化时间会被拉得很长，甚至永远追不上。

第二种是消费逻辑退化型积压。代码没改，流量没变，但Lag就是慢慢涨。这种情况最隐蔽，也最危险。常见原因包括：下游数据库连接池被打满导致每条消息处理时间从5ms涨到200ms；某个依赖的第三方接口超时重试；Consumer所在机器CPU被其他进程抢占。这种积压不会自己恢复，必须找到根因。

第三种是Consumer Group失衡型积压。明明有10个Consumer实例，但只有2个在干活，另外8个分不到分区。这通常是因为分区数少于消费者数，或者消费者加入组的时间差导致Rebalance反复触发。这种积压的特点是Lag分布极不均匀，有的分区Lag为零，有的分区Lag爆表。

1.2 用Lag曲线形态快速定位方向

我习惯先看Lag曲线的形状，这比看绝对值有用得多。

这个表不是绝对的，但能帮你在五分钟内缩小排查范围。我一般会同时打开Consumer的GC日志和线程堆栈，如果看到大量线程卡在socketRead或者数据库连接等待上，基本就能定位到方向。

1.3 紧急止血与根治手段

积压发生了，业务方在催，你不可能慢慢排查。我的做法是分两步走：先止血，再根治。

止血阶段，最直接的办法是临时扩容Consumer实例。但这里有个坑：如果你的Topic分区数是12，你扩容到20个Consumer，多出来的8个是分不到分区的，纯属浪费资源。所以扩容前一定要确认分区数。如果分区数不够，可以临时用Kafka的动态分区扩容命令增加分区，但要注意增加分区会触发Rebalance，而且如果消息有Key，分区数变了会导致Key的路由发生变化，可能影响业务逻辑。

另一个止血手段是跳过非关键消息。如果积压的消息里有一部分是可以丢弃的（比如日志类、埋点类），可以临时修改Consumer逻辑，用seek()方法跳过一部分Offset。但这个方法极其危险，一定要确认业务能接受数据丢失。

根治阶段，核心是提升消费吞吐量。我总结了几条经验：

• 批量消费：把max.poll.records从默认的500调大到2000甚至5000，配合手动提交Offset，能显著提升吞吐。但要注意单批处理时间不能超过max.poll.interval.ms，否则会触发Rebalance。
• 异步处理：把消息拉取和处理解耦，拉取线程只负责拉，处理线程池负责消费。但这样会带来Offset提交的复杂度，需要自己维护"已处理完成"的位点。
• 并行消费：在Consumer内部用线程池并行处理消息，但同样要解决Offset提交的顺序问题。我一般建议用"分区内有序、分区间并行"的策略，即每个分区一个处理线程，这样既能并行又不会乱序。

注意：异步处理和并行消费都会让Offset提交变得复杂，如果处理失败需要重试，一定要设计好重试队列和死信队列，否则消息丢了都不知道。

2. 消息过期：Kafka不负责"过期"，但你要负责"清理"

Kafka本身没有"消息过期"这个概念。消息是否过期，完全取决于你的业务逻辑和Consumer怎么处理。但实际场景中，我们确实需要处理"过期消息"——比如一条订单超时未支付的消息，过了30分钟就没有处理价值了。

2.1 消息过期的两种业务场景

场景一：消息本身有TTL。比如推送类消息，用户已经下线了，这条消息再推就没意义。这种场景下，消息的"过期时间"是写在消息体里的，Consumer拉取后先判断是否过期，过期就直接丢弃并提交Offset。

场景二：消息积压导致"被动过期"。比如积压了三个小时，Consumer终于消费到了，但业务上这些消息已经失效了。这种场景下，Consumer需要根据消息的时间戳来判断是否跳过。

两种场景的处理逻辑不同，但核心思路一致：在Consumer端做过滤，而不是指望Kafka帮你清理。

2.2 用时间戳过滤过期消息的实操

Kafka的消息里自带timestamp字段，可以在Consumer端直接读取。下面是一段Java代码示例：

@KafkaListener(topics = "order-topic", groupId = "order-group") public void consume(ConsumerRecord<String, String> record) { long messageTime = record.timestamp(); long now = System.currentTimeMillis(); long expireThreshold = 30 * 60 * 1000L; // 30分钟 if (now - messageTime > expireThreshold) { // 消息已过期，直接跳过 log.warn("消息已过期，跳过处理。offset={}, timestamp={}", record.offset(), messageTime); return; } // 正常处理逻辑 processOrder(record.value()); }

这段代码看起来简单，但有几个细节要注意：

• timestamp的类型：Kafka消息的timestamp有两种，CreateTime和LogAppendTime。CreateTime是生产者写入时的时间，LogAppendTime是Broker追加到日志的时间。如果生产者机器时钟不准，CreateTime可能偏差很大。我一般建议用LogAppendTime，但需要在Topic级别配置message.timestamp.type=LogAppendTime。
• 时钟同步：Consumer所在机器的时钟必须和Broker同步，否则now - messageTime的计算会出错。生产环境一定要开NTP。
• 过期消息的Offset提交：跳过消息后，Offset还是要提交的，否则下次还会拉到同一条消息。如果用的是自动提交，跳过就跳过了；如果是手动提交，记得在跳过逻辑后调用commitSync。

2.3 用死信队列兜底无法处理的消息

有些消息不是"过期"，而是"处理失败"。比如格式不对、依赖的数据不存在、下游接口返回异常。这种消息如果一直重试，会阻塞后面的消息。我的做法是引入死信队列（DLQ）。

具体流程是：Consumer处理失败后，把消息转发到一个专门的DLQ Topic，然后提交原Topic的Offset。DLQ Topic的消息可以由另一个Consumer慢慢处理，或者人工介入。

@KafkaListener(topics = "order-topic", groupId = "order-group") public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) { try { processOrder(record.value()); ack.acknowledge(); } catch (RetryableException e) { // 可重试异常，抛出后让Kafka重新投递 throw e; } catch (FatalException e) { // 不可重试异常，转入死信队列 kafkaTemplate.send("order-dlq-topic", record.key(), record.value()); ack.acknowledge(); log.error("消息转入死信队列，offset={}", record.offset(), e); } }

这里的关键是区分可重试异常和不可重试异常。网络超时、数据库连接失败属于可重试，格式错误、数据不存在属于不可重试。如果分不清，可以统一先重试三次，三次都失败再转入DLQ。

注意：死信队列的Topic也要设置合理的保留时间，否则DLQ也会积压。我一般设置DLQ的retention.ms为7天，给人工处理留足时间。

3. 读写分离在Kafka场景下的真实含义

"读写分离"这个词在数据库领域很常见，但在Kafka场景下，它的含义完全不同。Kafka的读写分离不是主从分离，而是生产与消费的隔离、不同消费组的隔离、关键业务与非关键业务的隔离。

3.1 为什么Kafka需要"读写分离"

Kafka的Topic可以被多个Consumer Group订阅，每个Group独立维护自己的Offset。这本身就是一种隔离。但实际场景中，我们还需要更细粒度的隔离。

举个例子：一个订单Topic，既需要被风控系统消费（要求低延迟），又需要被数据仓库消费（要求高吞吐，可以容忍延迟）。如果两个Consumer在同一个Group里，风控系统会被数据仓库的消费速度拖累。所以必须分成两个Group。

但分成两个Group后，又带来新问题：Broker的负载增加了。因为每个Group都要拉取全量消息，网络带宽和磁盘IO都会翻倍。这时候就需要考虑"读写分离"的另一种形式：用不同的Topic承载不同的消费需求。

3.2 多Group隔离与资源竞争

我见过一个典型的坑：某团队把风控和数据仓库放在同一个Consumer Group里，结果数据仓库做全量回溯时，风控的Lag飙升，导致风控规则延迟生效，差点出事故。

正确的做法是：

• 风控Group：独立Group，配置较小的max.poll.records（比如100），保证低延迟。
• 数仓Group：独立Group，配置较大的max.poll.records（比如5000），追求吞吐。
• 监控Group：独立Group，只消费少量消息做健康检查。

但这样做的代价是Broker要服务三倍的Consumer请求。如果Topic的流量很大，Broker的network线程可能成为瓶颈。这时候可以考虑用MirrorMaker或者自己写转发程序，把消息复制到不同的Topic，让不同的Consumer订阅不同的Topic。

3.3 用Topic复制实现彻底的读写分离

如果流量确实很大，我建议用Topic复制的方式。具体做法是：原始Topic只负责接收生产消息，然后用一个转发程序（可以是Kafka Streams、Flink，或者自己写的Consumer-Producer）把消息复制到多个下游Topic。

// 转发程序的核心逻辑 @KafkaListener(topics = "order-raw-topic", groupId = "forward-group") public void forward(ConsumerRecord<String, String> record) { // 转发到风控Topic kafkaTemplate.send("order-risk-topic", record.key(), record.value()); // 转发到数仓Topic kafkaTemplate.send("order-dw-topic", record.key(), record.value()); }

这个方案的好处是：风控Consumer只订阅order-risk-topic，数仓Consumer只订阅order-dw-topic，两者完全隔离。而且可以针对不同的Topic设置不同的保留时间、分区数、压缩策略。

但代价是：消息延迟增加了。因为多了一次转发，端到端延迟至少增加几十毫秒。如果业务对延迟极其敏感，这个方案就不合适。

注意：转发程序本身也可能成为瓶颈。如果原始Topic的流量是每秒十万条，转发程序需要能扛住这个量级。我一般建议用Flink或者Kafka Streams来做转发，它们天生支持水平扩展和Exactly-Once语义。

4. 一致性保障：从生产者到消费者的全链路思考

Kafka的"一致性"是个被过度简化的话题。很多人以为设置acks=all就万事大吉了，实际上从生产者到Broker再到消费者，每个环节都有一致性陷阱。

4.1 生产者端：acks与重试的权衡

生产者的acks参数有三个值：

• acks=0：生产者不等待Broker确认，直接发送下一条。速度最快，但消息可能丢失。
• acks=1：生产者等待Leader副本确认。如果Leader挂了且数据还没同步到Follower，消息会丢失。
• acks=all：生产者等待所有ISR副本确认。最安全，但延迟最高。

我一般建议核心业务用acks=all，日志类业务用acks=1。但acks=all也不是万无一失，如果ISR只剩Leader一个副本，acks=all就退化成acks=1了。所以还需要配合min.insync.replicas参数，确保至少有两个副本在ISR里。

# 生产者配置 acks=all retries=Integer.MAX_VALUE max.in.flight.requests.per.connection=1 # Broker配置 min.insync.replicas=2 unclean.leader.election.enable=false

这里有个关键点：max.in.flight.requests.per.connection=1。如果这个值大于1，且开启了重试，消息的顺序可能会乱。比如第一批消息发送失败，第二批消息发送成功，然后第一批重试成功，最终Broker上的顺序就反了。所以如果业务要求严格有序，这个值必须设为1。

4.2 Broker端：ISR机制与Unclean选举

Kafka的ISR（In-Sync Replicas）机制是保证一致性的核心。ISR里的副本都和Leader保持同步，只有ISR里的副本才有资格被选为新Leader。

但有一个危险参数：unclean.leader.election.enable。如果设为true，当ISR里所有副本都挂了，Kafka会允许非ISR副本成为Leader。这意味着数据可能丢失，但可用性提高了。我一般建议核心业务设为false，宁可不可用也不能丢数据。

另一个参数是replica.lag.time.max.ms，默认是30秒。如果Follower超过这个时间没追上Leader，就会被踢出ISR。如果这个值设得太大，ISR里可能有严重滞后的副本，acks=all的延迟会很高。如果设得太小，副本频繁进出ISR，会导致频繁的Leader选举。我一般建议根据实际网络延迟来调整，通常10到30秒比较合理。

4.3 消费者端：Offset提交与幂等处理

消费者端的一致性问题主要是重复消费。Kafka只能保证At Least Once，不能保证Exactly Once（除非用事务）。所以消费者必须做幂等处理。

幂等处理的常见方案：

• 数据库唯一键：把消息的业务ID作为唯一键，插入时如果冲突就忽略。
• Redis去重：用消息ID作为Key，处理前先查Redis，处理完写入Redis，设置合理的过期时间。
• 状态机：如果业务本身有状态流转，可以用状态机来保证幂等。比如订单状态从"待支付"到"已支付"，如果已经是"已支付"就忽略。

// 用Redis做幂等处理的示例 public void processOrder(String orderId, String messageId) { Boolean success = redisTemplate.opsForValue() .setIfAbsent("msg:" + messageId, "1", 24, TimeUnit.HOURS); if (Boolean.FALSE.equals(success)) { log.info("消息已处理，跳过。messageId={}", messageId); return; } // 正常处理逻辑 doProcessOrder(orderId); }

这里有个坑：Redis去重和业务处理不在同一个事务里。如果Redis写入成功但业务处理失败，下次消息重试时会被Redis拦截，导致业务永远处理不了。所以Redis的Key过期时间要足够长，或者用"先处理业务再写Redis"的顺序，但这样又可能重复处理。我一般建议用数据库唯一键，因为数据库事务能保证原子性。

4.4 端到端Exactly-Once的可行方案

Kafka 0.11之后引入了事务机制，可以实现端到端的Exactly-Once。但配置复杂，性能损耗也大。我一般只在金融级场景才建议用。

事务的核心配置：

# 生产者配置 enable.idempotence=true transactional.id=my-transactional-id # 消费者配置 isolation.level=read_committed

生产者的代码需要改成事务模式：

producer.initTransactions(); try { producer.beginTransaction(); producer.send(new ProducerRecord<>("topic", "key", "value")); producer.sendOffsetsToTransaction(offsets, "group-id"); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); }

这个方案能保证消息和Offset在同一个事务里提交，要么都成功，要么都失败。但代价是：吞吐量下降明显，因为每条消息都要等事务提交。我实测下来，事务模式的吞吐量大概是非事务模式的三分之一。

所以我的建议是：大部分场景用At Least Once + 幂等处理就够了，只有对数据一致性要求极高的场景才用事务。

5. 几个我踩过的坑和对应的解法

5.1 Rebalance导致的重复消费

有一次，Consumer Group频繁Rebalance，每次Rebalance后都会重复消费一批消息。排查后发现是max.poll.interval.ms设得太小，Consumer处理一批消息的时间超过了这个值，被Coordinator踢出Group，触发Rebalance。

解法：把max.poll.interval.ms从默认的5分钟调大到10分钟，同时把max.poll.records从500调小到200，确保单批处理时间不超过阈值。

5.2 消息顺序错乱

有一次，业务方反馈消息顺序乱了。排查后发现是生产者开启了重试且max.in.flight.requests.per.connection大于1。第一批消息发送失败，第二批成功，然后第一批重试成功，顺序就反了。

解法：把max.in.flight.requests.per.connection设为1，牺牲一点吞吐量换取顺序保证。如果业务对顺序要求极高，还可以用Key来保证同一Key的消息落到同一分区，这样分区内有序。

5.3 消费者Lag监控缺失

有一次，消息积压了三个小时才被发现，因为没有配置Lag监控。后来我加上了Prometheus + Grafana的监控，对每个Consumer Group的Lag设置告警阈值。

监控配置的核心指标：

• kafka_consumergroup_lag：Consumer Group的Lag值。
• kafka_consumergroup_lag_sum：所有分区的Lag总和。
• kafka_topic_partition_current_offset：当前Offset。
• kafka_topic_partition_oldest_offset：最老的Offset。

告警规则我一般设两级：Lag超过10万发警告，超过100万发严重告警。同时要结合Lag的增长速率，如果Lag在5分钟内增长了50%，即使绝对值不大也要告警。

5.4 磁盘写满导致的Broker宕机

有一次，Broker磁盘写满，导致所有写入失败。排查后发现是日志保留时间设得太长，而且没有配置磁盘配额。

解法：设置log.retention.hours=72（3天），log.retention.bytes=107374182400（100GB），并且开启log.cleanup.policy=delete。同时要监控磁盘使用率，超过80%就告警。

注意：如果Topic有多个，log.retention.bytes是每个分区的限制，不是整个Broker的限制。所以实际磁盘使用量是分区数乘以这个值。配置时要算清楚。

6. 一些配置参数的经验值

最后分享一些我常用的配置参数，都是实际跑过生产环境的经验值，供参考。

这些值不是绝对的，要根据实际业务场景调整。比如max.poll.records，如果单条消息处理要1秒，那设2000就会导致单批处理时间2000秒，远超max.poll.interval.ms，必然触发Rebalance。所以调参前一定要先算清楚单批处理时间。

我在实际使用中发现，大部分Kafka问题都不是Kafka本身的问题，而是配置和使用方式的问题。把上面这些点都考虑到，基本能避开80%的坑。剩下的20%，就要靠监控和日志来发现了。