企业官网建设流程全解析

1. 这不是理论考试，是生产环境里每天要签的“生死状”

Flink 的 Exactly Once 和 At Least Once，这两个词在面试题里出现频率高得离谱，但真正让一线开发者头皮发紧的，从来不是背定义，而是某天凌晨三点收到告警：下游数据表里突然多出十万条重复订单，或者关键指标统计结果比昨天少了整整 17%。这时候翻文档、查源码、看社区 issue，最后发现根子就卡在“我当初以为配对了 checkpoint，就等于万事大吉”——而实际上，Flink 的语义保障从来不是开个开关就能自动生效的魔法，它是一整套环环相扣的契约，从 Source 端读取、算子内部状态更新、到 Sink 端写入，每一环都必须主动签字画押，缺一不可。

Exactly Once 不是 Flink 给你的承诺书，而是你和 Flink 共同签署的一份操作协议。它要求：一次处理，一次生效，不多不少。At Least Once 则是它的“保守兄弟”：宁可多写几次，也绝不错过一次。听起来 At Least Once 更安全？错。在金融对账、库存扣减、实时计费这类场景里，“多写一次”可能直接触发风控熔断，或者让客户收到两笔退款。所以这不是语义优劣的选择题，而是业务容忍度的判断题——你得先问清楚业务方：“如果同一笔支付记录被计算两次，系统能自动识别并去重吗？还是说，只要发生，就必须人工介入回滚？”这个问题的答案，直接决定你该走哪条路。

这篇文章不讲教科书定义，也不堆砌源码片段。我用过去三年在三个不同规模实时数仓项目里踩过的坑、调过的参数、压测过的阈值，把 Exactly Once 和 At Least Once 拆解成一张可执行的操作清单。你会看到：为什么 Kafka Source 配置enable.auto.commit=false是铁律；为什么 MySQL Sink 用INSERT IGNORE会悄悄破坏 Exactly Once；为什么 checkpoint 间隔设成 30 秒，在高峰期反而导致状态后移超时失败；甚至包括一个被很多教程忽略的细节——Flink 1.15+ 版本里，TwoPhaseCommitSinkFunction的transactionTimeout默认值从 60 分钟降为 10 分钟，这个改动让某次大促期间的事务超时率飙升了 400%。所有内容，都来自真实日志、监控截图和线上回滚记录。

2. 语义保障的本质：不是“有没有”，而是“谁来负责”

2.1 Exactly Once 的三道生死门：Source、Operator、Sink 必须全部通关

很多人误以为只要开了 checkpoint，Flink 就天然支持 Exactly Once。这是最大的认知陷阱。Flink 的 checkpoint 机制本身只保证Operator 状态的一致性恢复，它管不了数据从哪来、往哪去。真正的端到端 Exactly Once，必须同时满足三个条件：

• Source 端可重放：当任务失败重启时，Source 必须能从上次消费的精确位置（offset）继续读，不能丢也不能重。Kafka 是典型支持者，但前提是消费者组配置enable.auto.commit=false，且 Flink 使用KafkaConsumer的setStartFromSpecificOffsets或setStartFromGroupOffsets方式初始化 offset。如果用了setStartFromEarliest，那每次重启都从头开始，语义直接退化为 At Most Once。

• Operator 状态可快照：这是 Flink 自身能力。Flink 通过异步 barrier 对齐机制，确保所有算子在同一个 checkpoint ID 下保存的状态是全局一致的。但这里有个隐藏雷区：如果你在map函数里手动 new 了一个 HashMap 存临时数据，这个 HashMap 不在 Flink 管理的状态后端里，checkpoint 时根本不会序列化它。任务重启后，这个 Map 就是空的——状态丢失，语义崩塌。正确做法是使用ValueState、ListState等 Flink 提供的状态接口。

• Sink 端幂等或事务：这是最容易翻车的一环。Flink 无法替你保证写入外部系统的操作是原子的。比如向 MySQL 写数据，如果用JDBCOutputFormat直接执行INSERT INTO，一旦任务在写入中途失败，checkpoint 恢复后会重放这一批数据，导致重复插入。解决方案只有两个：一是 Sink 实现幂等（如INSERT IGNORE或ON DUPLICATE KEY UPDATE），二是启用两阶段提交（2PC），让写入操作纳入 Flink 的 checkpoint 生命周期管理。后者才是严格意义上的 Exactly Once，前者只是“看起来像”。

提示：幂等写入 ≠ Exactly Once。INSERT IGNORE在主键冲突时跳过，看似解决了重复问题，但如果业务逻辑需要“首次写入才触发下游通知”，那么第二次IGNORE就意味着通知永远丢失。这已经违背了“一次处理，一次生效”的核心原则。

2.2 At Least Once 的“安全假象”：它用资源换确定性

At Least Once 的设计哲学很直白：宁可多做，不可不做。它只要求 Source 可重放、Operator 状态可快照，对 Sink 的要求大幅降低——哪怕 Sink 写入失败，Flink 也会在 checkpoint 恢复后重试。这种宽松换来的是更高的吞吐和更低的延迟，因为不需要等待 Sink 的事务提交确认。

但它付出的代价是显性的：资源消耗翻倍风险。我们曾在一个日均 20 亿条日志的清洗链路中启用 At Least Once，观察到两个现象：一是 Flink Web UI 中numRecordsInPerSecond指标在高峰期出现明显毛刺，峰值流量比平时高出 30%；二是下游 Kafka topic 的堆积量在任务重启后激增，因为重放的数据流和新数据流叠加涌向下游。根本原因在于：At Least Once 下，Flink 不会主动丢弃已发送但未确认的数据。当 Sink 因网络抖动返回TimeoutException时，Flink 认为“可能成功了”，于是保留这批数据的引用，直到下一次 checkpoint 成功——这期间，这些数据既没被确认，也没被丢弃，处于一种“薛定谔的已发送”状态，持续占用内存和网络带宽。

更隐蔽的风险在于下游系统压力失真。如果你的监控体系只看 Flink 的numRecordsOutPerSecond，会误以为流量平稳；但实际下游 Kafka broker 的RequestHandlerAvgIdlePercent可能已跌破 30%，因为同一份数据被反复推送。我们在某次压测中发现，当 At Least Once 链路遭遇 5% 的 Sink 失败率时，下游 Kafka 的请求队列长度平均增长 2.8 倍，而 Flink 自身的 CPU 使用率反而下降——因为大量线程阻塞在重试等待上。

2.3 语义选择不是技术决策，而是业务 SLA 的翻译

把技术术语翻译成业务语言，才能做出正确选择。我们整理了一张决策对照表，基于过去项目的真实案例：

这张表的核心逻辑是：语义选择的权重，永远由业务错误成本决定，而非技术实现难度。当“少算一笔”和“多算一笔”的业务影响完全不对称时（比如财务场景），必须选 Exactly Once；当系统天然具备容错能力（如归因模型、推荐排序），At Least Once 的资源效率优势就值得采纳。

3. Exactly Once 落地实操：从配置到压测的完整闭环

3.1 Kafka Source 端：三个配置决定语义生死

Kafka 是 Flink 最常用的 Source，但默认配置下，它连 At Least Once 都无法稳定保障。我们必须手动关闭 Kafka 自身的 offset 提交机制，并将 offset 管理权完全交给 Flink。以下是生产环境验证过的最小必要配置：

Properties props = new Properties(); props.setProperty("bootstrap.servers", "kafka-broker:9092"); props.setProperty("group.id", "flink-consumer-group"); // 【致命配置】必须关闭 Kafka 自动提交，否则 Flink checkpoint 保存的 offset 和 Kafka 实际 offset 会脱节 props.setProperty("enable.auto.commit", "false"); // 【关键配置】设置 session.timeout.ms，避免消费者被误踢出组导致 rebalance props.setProperty("session.timeout.ms", "30000"); // 【增强配置】设置 max.poll.records，防止单次拉取过多数据导致处理超时 props.setProperty("max.poll.records", "500"); FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>( "input-topic", new SimpleStringSchema(), props ); // 【核心配置】指定从 group offset 开始消费，确保重启时读取 checkpoint 保存的位置 consumer.setStartFromGroupOffsets(); // 【可选但强烈推荐】设置 checkpoint 间隔，与 Kafka fetch timeout 匹配 env.enableCheckpointing(60000); // 60秒 checkpoint

为什么enable.auto.commit=false如此关键？我们曾在线上遇到一个经典故障：某天凌晨 Kafka 集群短暂抖动，部分 partition leader 切换。Flink 任务因网络超时触发 checkpoint 失败，自动重启。重启后，Kafka Consumer 因enable.auto.commit=true，在 rebalance 完成前已将 offset 提交到旧 leader，而 Flink 从 checkpoint 恢复时，却试图从新 leader 的旧 offset 读取——结果就是大量数据重复消费。关闭自动提交后，Flink 在每次 checkpoint 成功时，同步将当前消费的 offset 作为状态的一部分写入 backend，重启时统一恢复，彻底规避 offset 错位。

注意：setStartFromGroupOffsets()并非万能。如果 Kafka topic 的 retention 时间短于 Flink 任务的 checkpoint 间隔，可能出现“想从 offset X 读，但 Kafka 已删除该 offset 数据”的情况。此时需配合setStartFromLatest()或setStartFromTimestamp(System.currentTimeMillis() - 3600000)设置安全起始点。

3.2 Operator 状态管理：避开“手写状态”的三大陷阱

Flink 的状态后端（RocksDB 或 Heap）负责持久化算子状态，但开发者常犯的错误是混淆“Java 对象”和“Flink 状态”。以下是三个高频踩坑点及修复方案：

陷阱一：在 open() 方法中初始化非状态对象

public class BadMapper extends RichMapFunction<String, String> { private HashMap<String, Integer> cache; // ❌ 错误：HashMap 不受 Flink 管理 @Override public void open(Configuration parameters) { cache = new HashMap<>(); // 重启后此 map 为空，状态丢失 } @Override public String map(String value) { String key = parseKey(value); cache.put(key, cache.getOrDefault(key, 0) + 1); // 状态仅存在于内存 return value; } }

修复方案：改用 Flink 状态接口，让状态纳入 checkpoint 生命周期：

public class GoodMapper extends RichFlatMapFunction<String, String> { private transient ValueState<Integer> counterState; // ✅ 正确：Flink 管理的状态 @Override public void open(Configuration parameters) { ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("counter", Types.INT); counterState = getRuntimeContext().getState(descriptor); } @Override public void flatMap(String value, Collector<String> out) throws Exception { String key = parseKey(value); // 设置 key，使状态按 key 分片 getRuntimeContext().getPartitionedState( new ValueStateDescriptor<>("counter", Types.INT) ).update(counterState.value() == null ? 1 : counterState.value() + 1); out.collect(value); } }

陷阱二：状态过大导致 checkpoint 超时RocksDB 状态后端虽支持大状态，但单个 Key 的状态值若超过 1MB，序列化/反序列化耗时会急剧上升。我们曾有一个用户画像聚合任务，状态中缓存了用户最近 1000 次点击的 URL 列表，单个用户状态平均达 1.2MB。结果是 checkpoint 平均耗时从 800ms 暴涨至 4.2s，频繁触发CheckpointDeclinedException。

修复方案：对大状态进行分片或压缩：

• 分片：将长列表拆分为多个 Key，如user_123_clicks_001,user_123_clicks_002；
• 压缩：对 URL 列表使用 LZ4 压缩后再存入ValueState<byte[]>；
• 采样：业务允许时，只保留最近 100 次点击，用LinkedBlockingQueue实现 FIFO 缓存。

陷阱三：状态 TTL 设置不当引发数据泄露Flink 状态支持 TTL（Time-To-Live），但StateTtlConfig的cleanupFullSnapshot()参数常被忽略。该参数控制是否在全量 checkpoint 时清理过期状态。若设为false（默认），过期状态仍会写入 checkpoint 文件，导致磁盘空间持续增长。

修复方案：显式开启全量清理：

StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.days(7)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .cleanupFullSnapshot() // ✅ 强制在全量 checkpoint 时清理 .build();

3.3 Sink 端 Exactly Once 实现：两阶段提交（2PC）的深度解析

Flink 官方提供的TwoPhaseCommitSinkFunction是实现 Exactly Once Sink 的标准范式，但其内部机制常被简化为“开启事务”四个字。实际上，它是一套精密的四步协同协议：

Step 1：预提交（Pre-commit）当 checkpoint barrier 到达 Sink 算子时，Flink 触发snapshotState()方法。此时，Sink 不执行真实写入，而是：

• 创建一个临时事务句柄（如 MySQL 的XA START 'tx_id'）；
• 将待写入数据暂存到该事务的私有上下文中（非真实数据库表）；
• 返回一个可序列化的事务快照（包含 tx_id、数据批次 ID、写入位置等）给 Flink。

Step 2：checkpoint 确认（Checkpoint Complete）Flink 收到所有算子的快照后，向 JobManager 发送确认。此时，notifyCheckpointComplete()方法被调用。Sink 执行：

• 提交临时事务（如XA COMMIT 'tx_id'），将数据真正刷入数据库；
• 清理本地缓存，释放资源。

Step 3：失败恢复（Failure Recovery）若任务在 Step 1 和 Step 2 之间失败，重启后 Flink 会调用restoreState()，传入之前保存的事务快照。Sink 根据快照中的 tx_id：

• 查询数据库，确认该事务是否已提交；
• 若未提交，则执行XA ROLLBACK清理脏数据；
• 若已提交，则跳过，避免重复提交。

Step 4：超时兜底（Timeout Handling）这是最易被忽视的环节。Flink 1.15+ 将transactionTimeout默认值从 3600000ms（60分钟）改为 600000ms（10分钟）。这意味着：如果一个事务在 10 分钟内未收到notifyCheckpointComplete()，Flink 会主动调用abort()方法强制回滚。

实操参数调优表：

我们曾因未调整transactionTimeout，在一次数据库主从切换期间，导致 17% 的事务被误判为超时而回滚，最终数据缺失。将该值提升至 1800000ms 后，同类故障归零。

3.4 端到端 Exactly Once 验证：三类必做压测

配置完成不等于语义落地。必须通过以下三类压测验证：

压测一：断点续传验证

• 步骤：启动任务 → 消费 1000 条数据 → 手动 kill TaskManager → 等待重启 → 继续消费 1000 条；
• 验证点：下游数据库总记录数 = 2000，且无重复主键冲突；
• 工具：使用SELECT COUNT(*), COUNT(DISTINCT id) FROM table对比。

压测二：高并发写入验证

• 步骤：模拟 1000 QPS 持续写入 1 小时 → 在第 30 分钟随机 kill JobManager；
• 验证点：检查 checkpoint 日志，确认notifyCheckpointComplete调用次数与snapshotState次数一致；下游数据无跳跃或重复。

压测三：网络分区验证

• 步骤：使用tc命令在 Flink TaskManager 和 MySQL 之间注入 500ms 网络延迟 + 5% 丢包；
• 验证点：Flink Web UI 中checkpointStatus显示IN_PROGRESS时长不超过transactionTimeout；任务不持续重启。

实操心得：压测时务必开启 Flink 的详细日志级别（log4j.logger.org.apache.flink=DEBUG），重点关注TwoPhaseCommitSinkFunction类的日志。我们曾通过日志发现，某次故障并非事务超时，而是 MySQL 的wait_timeout（默认 8 小时）被提前触发，导致连接被服务端关闭，而 Sink 未及时捕获SQLException，误判为网络超时。解决方案是在 JDBC URL 中添加autoReconnect=true&failOverReadOnly=false。

4. At Least Once 场景下的“可控重复”实践指南

4.1 主动设计幂等写入：让重复变得可预测、可管理

At Least Once 的核心优势是低延迟，但必须将“重复”转化为“可控”。我们不追求消灭重复，而是让重复发生在业务可接受的边界内。以下是三种经过生产验证的幂等模式：

模式一：业务主键 + UPSERT适用于有明确业务主键的场景（如订单 ID、用户 ID）。MySQL 示例：

INSERT INTO user_profile (user_id, last_login_time, profile_data) VALUES (?, ?, ?) ON DUPLICATE KEY UPDATE last_login_time = VALUES(last_login_time), profile_data = VALUES(profile_data);

关键点：ON DUPLICATE KEY UPDATE中的VALUES()函数确保新值覆盖旧值，而非累加。我们曾因误写last_login_time = last_login_time + VALUES(last_login_time)，导致时间戳翻倍，用户登录态异常。

模式二：版本号乐观锁适用于需保留历史变更的场景（如商品价格更新）。表结构增加version字段：

UPDATE product_price SET price = ?, version = version + 1 WHERE product_id = ? AND version = ?;

Flink Sink 在写入前，先查询当前version，写入时带上该version作为条件。若UPDATE返回影响行数为 0，说明已被其他任务更新，此时可选择丢弃或重试。我们在线上采用“重试 2 次 + 降级为覆盖写入”策略，确保最终一致性。

模式三：消息指纹去重适用于无业务主键，但数据本身可哈希的场景（如日志行、埋点事件）。步骤：

• 对原始数据计算 SHA-256 指纹（如SHA2(CONCAT(event_type, user_id, timestamp), 256)）；
• 将指纹作为 Redis 的 Key，设置过期时间（如 24 小时）；
• 写入前SETNX指纹 Key，成功则写入，失败则跳过。性能优化：使用 Redis Cluster 分片存储指纹，单节点压力过大时，将指纹哈希后路由到不同 slot。

注意：指纹去重的时效性取决于 Redis 过期时间。若业务要求“绝对不重复”，需将过期时间设为远大于数据最大乱序窗口（如 7 天），但这会带来 Redis 内存压力。我们通过监控redis_memory_used_bytes和redis_keyspace_hits，动态调整过期时间，在内存增长速率 > 5GB/天时，自动缩短为 48 小时。

4.2 At Least Once 下的资源监控黄金指标

选择 At Least Once 后，必须建立一套专属监控体系，因为传统指标会失真。以下是三个必须盯紧的核心指标：

指标一：重放率（Replay Ratio）计算公式：(numRecordsInPerSecond - numRecordsOutPerSecond) / numRecordsInPerSecond * 100%

• 正常值：0% ~ 5%（网络抖动、GC 暂停导致）；
• 预警值：> 10%（可能 Sink 性能瓶颈或下游拒绝）；
• 故障值：> 30%（需立即检查下游 Kafka backlog 或 DB 连接池）。

指标二：Checkpoint 间隔波动率计算公式：ABS(current_interval - baseline_interval) / baseline_interval * 100%

• baseline_interval 为稳定期的平均 checkpoint 间隔（如 30s）；
• 波动率 > 50% 时，表明状态写入或 Sink 提交存在严重延迟，即使未报错，语义保障已弱化。

指标三：Sink 重试队列长度Flink 1.14+ 暴露了numPendingCheckpoints和numCompletedCheckpoints，但更关键的是自定义 Sink 的重试队列。我们在 Kafka Sink 中添加了retryQueueSize指标：

• 队列长度持续 > 1000：说明网络或下游 Kafka 不可用，重试积压；
• 队列长度周期性尖峰：表明下游存在间歇性限流（如 Kafka broker 的request_queue_size达上限）。

我们曾通过retryQueueSize发现，某次故障并非 Flink 问题，而是 Kafka 的replica.fetch.wait.max.ms参数过小（默认 500ms），导致 follower 副本拉取慢，broker 主动断开连接，触发 Flink 重试。将该参数调至 2000ms 后，重试率下降 92%。

4.3 从 At Least Once 平滑升级 Exactly Once 的路径

业务演进中，常需将 At Least Once 链路升级为 Exactly Once。强行切换会导致数据中断。我们总结出四步平滑迁移法：

Step 1：双写并行（Shadow Mode）
修改 Sink，同时向旧目标（At Least Once）和新目标（Exactly Once）写入相同数据。新目标使用独立 topic 或表，不参与线上业务。此阶段验证 Exactly Once 链路的稳定性，不影响现有业务。

Step 2：数据比对（Reconciliation）
开发比对脚本，定时扫描双写数据：

• 检查新旧目标的记录总数、主键分布、关键字段 MD5 值；
• 使用抽样比对（如每 1000 条抽 1 条）降低计算压力；
• 我们用 Spark SQL 实现，单次全量比对耗时 < 8 分钟。

Step 3：灰度切流（Canary Release）
选取 1% 的流量（如特定 user_id 段）切到 Exactly Once 链路，其余 99% 保持 At Least Once。监控灰度流量的延迟、错误率、下游数据质量。若 24 小时无异常，逐步扩大比例。

Step 4：全量切换与回滚预案
全量切换前，执行最终一致性校验。同时准备回滚 SQL：

-- 若 Exactly Once 链路异常，快速切回 At Least Once UPDATE flink_job_config SET sink_mode = 'AT_LEAST_ONCE' WHERE job_name = 'realtime-etl'; -- 并清空 Exactly Once 的临时事务表（如有） TRUNCATE TABLE flink_xa_transactions;

整个过程历时 72 小时，零业务中断。

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

5.1 “Exactly Once 为什么还是有重复？”—— 五层排查法

当监控发现下游数据重复，按以下顺序逐层排查，90% 的问题可在前三层定位：

真实案例：某电商订单链路出现 0.02% 重复，L1-L4 均正常。最终在 L5 发现：订单创建事件被同时写入order_raw和payment_event两个 topic，而两个 Flink 作业都订阅了order_raw，导致同一订单被处理两次。解决方案是引入事件中心，对事件类型做唯一路由。

5.2 “Checkpoint 总是失败”—— 高频原因与修复清单

Checkpoint 失败是 Exactly Once 的最大拦路虎。我们整理了生产环境中 Top 5 原因及修复方案：

原因一：状态后端写入超时（占比 42%）

• 现象：日志中出现Checkpoint failed due to timeout，RocksDB相关指标rocksdb_flush_pending持续 > 0；
• 根因：RocksDB 的write_buffer_size过小（默认 64MB），在高写入场景下频繁 flush，阻塞 checkpoint；
• 修复：增大write_buffer_size至 256MB，并启用level_compaction_dynamic_level_bytes=true。

原因二：网络屏障（barrier）对齐超时（占比 28%）

• 现象：CheckpointCoordinator日志显示Barrier has not aligned for x seconds；
• 根因：某个 subtask 处理速度远慢于其他 subtask（数据倾斜），导致 barrier 无法对齐；
• 修复：
  1. 使用Flink Web UI -> Task Managers -> Metrics查看各 subtask 的busyTimeMsPerSecond；
  2. 对慢 subtask 进行keyBy重分区，如keyBy(x -> x.hashCode() % 100)打散热点 key；
  3. 增加execution.checkpointing.alignment-timeout至 60000ms（谨慎使用，治标不治本）。

原因三：JVM GC 导致线程暂停（占比 15%）

• 现象：Checkpoint 失败时间点与 GC 日志中Full GC时间高度重合；
• 根因：Heap 设置过大（> 8GB）导致 CMS GC 停顿时间过长；
• 修复：切换至 G1 GC，设置-XX:+UseG1GC -XX:MaxGCPauseMillis=200，并限制 Heap ≤ 6GB。

原因四：外部系统响应慢（占比 10%）

• 现象：notifyCheckpointComplete耗时 >transactionTimeout；
• 根因：MySQL 的innodb_log_file_size过小，大事务写入 redo log 缓慢；
• 修复：将innodb_log_file_size从 48MB 提升至 256MB，并重启 MySQL。

原因五：状态序列化失败（占比 5%）

• 现象：日志中Could not serialize state，伴随NotSerializableException；
• 根因：自定义 Function 中引用了不可序列化的对象（如ThreadLocal、Connection）；
• 修复：将非序列化对象声明为transient，并在open()方法中重新初始化。

5.3 “At Least Once 下游堆积”—— 流量整形实战技巧

当 At Least Once 链路下游 Kafka 出现堆积，单纯扩容 Flink 并非最优解。我们采用“三级流量整形”策略：

Level 1：Flink 内部反压（Backpressure）

• 启用taskmanager.network.memory.fraction（默认 0.1），确保网络缓冲区充足；
• 监控backPressuredTimeMsPerSecond，若 > 1000ms，说明下游已饱和，Flink 自动降速。

Level 2：Kafka Producer 限流
在 Flink Kafka Sink 中配置：

# 控制每秒最多发送 1000 条，避免突发流量打爆下游 linger.ms=5 batch.size=16384 # 启用压缩，减少网络传输量 compression.type=lz4

Level 3：业务级降级（Circuit Breaker）
当 Kafka backlog > 100 万条时，触发降级：

• 自动将max.poll.records从 500 降至 100；
• 将fetch.max.wait.ms从 500 提升至 1000，延长拉取等待；
• 向监控系统发送ALERT: KAFKA_BACKLOG_HIGH事件，触发人工介入。

这套组合拳使某次大促期间的 Kafka 堆积峰值从 240 万条压降至 47 万条，且未影响上游数据采集。

6. 个人经验总结：语义保障没有银弹，只有权衡的艺术

我在三个不同行业的实时计算项目里，反复验证了一个朴素真理：Flink 的 Exactly Once 和 At Least Once，从来不是非此即彼的技术选项，而是业务复杂度、系统成本、运维能力三者博弈后的平衡点。第一次做金融风控时，我迷信 Exactly Once，把所有链路都配成 2PC，结果发现 TPS 不足预期的 60%，最终不得不将非核心维度的聚合降级为 At Least Once + 后