★ FEATURED ARTICLE 1. 为什么Kafka默认只能二选一至少一次与最多一次很多人刚接触Kafka时最先被安利的概念就是“它是个高吞吐、可持久化的消息队列”。等到真正做业务了第一个被问倒的问题往往是线上到底会不会丢消息、会不会重复消费如果去翻官方文档会看到三组词——at most once、at least once、exactly once也就是最多一次、至少一次、恰好一次。大部分初级工程师的第一反应是那我直接选exactly once不就行了但现实是Kafka在很长一段时间里默认配置只给你“至少一次”想要“恰好一次”必须靠一整套精心设计的机制去拼出来。先说清楚为什么默认只能是“至少一次”。一条消息从生产者到消费者中间要经历四个关键环节生产者把消息发到broker、broker落盘并同步副本、消费者拉取消息处理业务、消费者提交位移。任何一个环节都可能出岔子。生产者发送消息之后网络抖动没有收到broker的ack它重试一次这时候broker那边可能已经写成功了重试的结果就是同一条消息写了两遍下游消费到两张一模一样的订单。这是产生重复的经典路径。再看消费端假设消费者拿到消息、处理完业务之后还没来得及提交位移就宕机了重启之后位移还是旧值于是这条消息又被拉取一次业务也就重复执行了一遍。为什么会这样因为Kafka当初的设计目标就是“高性能、高吞吐”它把底层的存储和副本同步做得很极致但“不重不漏”这种分布式系统里的圣杯问题它不可能靠单点机制就解决。broker能承诺的是只要ISR里至少有一个副本活着消息就不会丢消费端能承诺的是只要你提交了位移下次就从位移1的位置继续读。但“消息不丢”和“业务不重复执行”之间隔着一整条网络的不确定性。这时候会有同学问那我用“最多一次”行不行最多一次的意思是消费者先提交位移再处理业务。这样即便宕机位移也已经在前面了重启用新位移继续读旧消息自然被跳过业务不会重复执行但代价是——如果处理过程中宕机那条消息对应的业务就彻底没做了。订单没创建、库存没扣减而且你根本不知道。所以默认情况下Kafka给的是“至少一次可能重复”这个组合。它保证了消息不丢但没保证业务不重。而真正业务系统最怕的其实是“重”因为重意味着幂等逻辑要自己写、对账要自己做、脏数据要自己清理。这也是为什么“恰好一次”的语义会成为中间件领域绕不开的话题。1.1 位移提交和消息落盘的矛盾拆开来看问题的本质是一个“两阶段提交”式的矛盾消息落盘和位移提交是两个独立的存储动作一个在broker端一个在消费端group协调器里两者之间没有强一致协议。你处理完业务再去提交位移可能重复你提交完位移再处理业务可能丢。这个矛盾是所有消息队列的通用难题。RabbitMQ的做法是手工ack说穿了也是把选择权交给业务方RocketMQ则推出了事务消息用半消息回查来解决Kafka的方式很不一样它没有在broker端做类似分布式事务的二阶段锁而是把问题拆成两半——写入时用幂等生产者解决重复提交时用事务把“写结果”和“提交位移”绑成一个原子单元。要理解后续的所有内容最好先记住这样一个判断Kafka的exactly once并不是让每条消息在整个链路里“恰好被处理一次”而是让“处理”这件事在业务结果上表现为恰好一次。这个区别非常关键后面所有设计都是围绕这一点展开的。1.2 “不重不漏”的成色真正缺的是跨端原子性再往深一层说Kafka能做的“不重不漏”是有边界的。它只能保证在Kafka内部从生产者写入到消费者读取这段路径上消息不会因为网络重试而重复也不会因为broker故障而丢失。但消息被消费之后业务代码往自己的数据库里写的那条SQLKafka管不着。你把订单写入MySQL、把库存扣减在Redis、把日志打到ES这些步骤如果中途失败Kafka没有任何机制能让它们回滚。所以说清楚边界很重要Kafka的恰好一次是一套“生产者幂等 事务提交 消费位移事务化”的组合拳它的落点是“Kafka消息流内的原子性”而不是“业务系统全链路的分布式事务”。搞清楚这个边界再去看那些卖力吹嘘“Kafka保证不重不漏”的文章你就知道哪些是靠谱的哪些是在偷换概念。2. 幂等生产者PID 序列号如何把重复请求压住幂等生产者是Kafka走向exactly once的第一步也是很多人容易忽略的一步。Kafka 0.11引入了这个能力解决的就是最经典的痛点生产者重试导致的重复写入。它的设计思路用一句话概括——给每一条消息发一个唯一的序列号broker端按序号去重。具体机制不复杂。生产者启动时会向broker申请一个PID也就是producer id这个ID对应用户无感知但它会伴随着生产者的整个生命周期。同时每个生产者实例会维护一个单调递增的sequence number。发往同一个分区的一批消息会带上这个PID和sequence。broker端为每个分区维护着一个已接收的最大序列号收到消息之后先比对如果新来的序列号比已经记录的还小说明是重试产生的重复消息直接拒绝。这个方案看起来简单但有几个细节很容易踩坑值得展开说。2.1 幂等机制原理不依赖业务层的兜底方案最值得说的就是这套幂等去重是Kafka服务端内部的它不需要业务方改任何接口、不需要在消息体里塞业务唯一键、不需要自己在消费端做去重表。它的去重粒度是“生产者实例级别 分区级别”只要你的序列号是连续递增的broker就能识别出哪些是重试导致的旧消息。举个例子。某个订单服务往users-topic的partition 0写消息连续写了1、2、3三条每条都带PID1001sequence分别是0、1、2。第三条发送时网络超时生产者重试重新发送sequence2的消息。这次重试到达broker时broker发现自己记录的最大序列号就是2于是识别出这是一条重复消息直接返回成功但不再落盘。这样消费者看到的还是1、2、3三条不会出现重复。这里有一个非常容易被误解的点很多人以为幂等生产者是Kafka在broker端做了一张全局去重表或者按消息内容算了个hash。实际上都不是它是依赖“同一个PID下序列号的连续性”来做判断的。如果消息乱序到达比如sequence3先到、sequence2后到在开启幂等的情况下broker会直接报错。所以Kafka要求生产者发往同一分区的消息必须是按顺序发送的这是幂等机制的前提也是为什么默认开启幂等时会限制max.in.flight.requests.per.connection不能超过5的原因。2.2 幂等边界为什么只能处理单分区内的重试幂等机制能解决的仅限于“同一个生产者实例、同一分区、因网络重试造成的重复”。它的边界非常清晰第一它管不了生产者重启。因为生产者重启之后会重新申请PID旧PID的序列号状态全部作废同一台机器上重发之前的消息broker会认为这是一个全新生产者的全新消息序列重复无法被识别。第二它管不了消费者处理阶段的重复。消费者拿到消息、处理业务、提交位移这个循环在幂等生产者层面完全不涉及。就算生产者保证了消息写入不重复消费端那边该重复消费还是重复消费。第三它管不了跨分区的事务性。如果一条业务数据要同时写到两个topic或者写到同一topic的两个分区分A成功、分B失败这个结果是不一致的幂等生产者无能为力。需要把多个分区的写入“绑定”成一个不可分割的操作——这就是事务API要做的事。但即便有这些边界幂等生产者依然是性价比最高的改动开启成本几乎为零只需要把enable.idempotence设为true同时acks设为all。用它来兜住重试导致的重复是所有Kafka生产环境里都应该做的基础配置没有理由不开。3. 事务API跨分区写入的原子性如何落地如果说幂等生产者解决的是“单分区不重复”那事务API解决的就是“跨分区原子性”。很多业务场景里一条消息要同时写两个topic或者一条消息写进去了但对应的日志topic没写进去这种情况在金融、交易、对账场景里是不可接受的。Kafka的事务API借鉴了数据库二阶段提交的思路又融入了自己的一套协调器机制。它允许生产者把多个分区的写入标记为同一个事务一旦commit消费者要么能看到全部写成功要么什么也看不到不存在中间状态。这套机制从0.11版本开始提供发展到现在已经相当成熟。3.1 事务协调器与transactional.id重启后还能续命的机制事务API的关键引入了一个新组件transaction coordinator也就是事务协调器。它本质上是一个特殊的broker角色由消费者的group协调器逻辑衍生而来。每个生产者发起事务时先向coordinator注册自己的transactional.idcoordinator为其分配一个PID并维护事务状态。这里有个非常巧妙的设计transactional.id是用户配置的字符串它和PID是绑定关系但PID会随着生产者重启改变。事务协调器会把transactional.id和最新的PID绑定起来当生产者重启之后用同一个transactional.id去申请事务coordinator会把它和新的PID关联。这解决了幂等生产者那部分“重启就失效”的痛点——幂等只能保证单实例生命周期内的去重而事务语义在跨越实例重启之后依然能被识别。但代价是transactional.id必须唯一。如果你两个生产者的实例配了同一个transactional.id那它们会互相抢占事务控制权轻则事务报错重则出现未决中断事务。我们后来遇到过线上两台服务配了相同transactional.id的情况表现就是频繁出现ProducerFencedException消费端看到一堆aborted事务排查了半天才定位到是配置冲突。3.2 提交/中止标记、LSO 与 read_committed 消费者的配合事务要真正对消费者产生影响光靠生产者commit还不行消费端也得配合。Kafka为事务设计了一套特殊的消费隔离机制核心概念有两个LSOlast stable offset和isolation.level。当生产者开始一个事务、向某个分区写入消息时这些消息在存储上被打上事务标记但普通消费者是读不到的。只有当事务被commit之后消息才对外可见。Kafka是怎么做到这一点的呢答案是broker为每个分区维护一个叫LSO的偏移量LSO之前的消息都是已稳定状态——要么是已提交事务的消息要么是非事务消息。事务进行中的消息它们的偏移量在LSO之后read_committed模式的消费者是检索不到的。消费端通过isolation.level参数控制读取行为read_uncommitted事务消息只要写入就可见和普通消息一样不区分提交与中止。read_committed只能读取已提交事务的消息未提交或已中止的事务消息会被过滤掉。这个过滤不是消费者自己做的而是broker在拉取时做了特殊处理。当一个事务abort时broker会写入一个abort marker标记这段消息作废read_committed消费者在拉取时会跳过这段数据但Offset并不会回退因为消息本身还在segment里。这也解释了为什么一边开启事务、一边用默认的read_uncommitted消费会出现读到“中间态数据”的情况——如果你业务的容忍度低这个参数必须改。3.3 一个事务流端到端的代码级走读用代码走一遍能看得更清楚。以下是一个典型的“读消息-处理-写多个topic”的事务流。// 初始化事务生产者 Properties props new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, payment-service-01); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); KafkaProducerString, String producer new KafkaProducer(props); producer.initTransactions(); try { // 业务操作把用户支付成功结果写入两个topic producer.beginTransaction(); producer.send(new ProducerRecord(order-topic, order-1001, paid)); producer.send(new ProducerRecord(wallet-topic, user-888, deduct-100)); // 在同一事务里把消费位移也提交到__consumer_offsets producer.sendOffsetsToTransaction(Map.of( new TopicPartition(order-input, 0), new OffsetAndMetadata(1024L) ), consumer-group-a); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); throw e; } finally { producer.close(); }注意这里最容易被忽略的一行sendOffsetsToTransaction。它把消费端的位移提交也纳入了同一个事务。这个操作是Kafka事务语义真正“闭环”的关键——业务结果写入和位移提交要么一起成功要么一起回滚。不这么做的话业务写入成功了位移没提交重启后会重复消费业务写入失败了位移却提交了重启后会丢消息两种情况都谈不上exactly once。顺带说一句事务消息在写入后、提交前这段窗口如果生产者进程突然崩溃事务就变成了未决状态。Kafka设置了一个超时时间transaction.timeout.ms默认60秒coordinator会在这个时间后自动中止未决事务。但如果你的事务里混合着外部系统操作这个60秒窗口的代价要提前评估——外部操作耗时长的话要调整timeout配置否则事务会莫名其妙被中断。4. 消费端EOS的关键读-处理-写循环才是真正的恰好一次很多人把Kafka的exactly once理解成“broker层面升级后的魔法”这是一个误区。翻遍官方文档你会发现EOS真正的实现场景是搭配Kafka Streams或者自研的读-处理-写模式。单靠生产者事务和消费者隔离其实还凑不出完整的“不重不漏”。我把这个逻辑拆开讲。假设有一条订单消息从input-topic进入你的业务要做的事是读消息、写MySQL扣库存、再给output-topic发一条结果。这里有两个独立的系统边界Kafka到MySQL以及MySQL到Kafka。Kafka的事务只覆盖了Kafka内部的写入MySQL那一步它管不了。所以要做到业务层面的恰好一次核心是把“消费位移”和“业务结果”放进同一个事务——业务结果写成功了位移才能提交业务结果没成功位移就回退重新消费。如果是纯Kafka内部的场景比如消息转换、聚合、过滤那Kafka Streams已经把这套链路封装好了。它能做到EOS靠的正是前面讲的事务机制。映射到用户视角就是每处理一条消息Kafka Streams先把中间数据写到changelog topic再把消费位移作为事务的一部分提交。整个过程对外表现为“处理结果和位移一起原子提交”从而实现了不重不漏。4.1 为什么消费位移和业务结果必须放在同一个事务里用一个很简单的例子解释这个必要性。消费者从Kafka拉了一条消息业务是把用户的余额从100改成50。如果先改MySQL成功了但在提交位移之前崩溃重启后位移是旧的这条消息会被再消费一次余额会从50变成0——这是重复执行事故。另一种顺序先提交位移再改MySQLMySQL更新失败重启后直接跳过这条消息余额没变、订单状态错了——这是丢失业务事故。两条路都是错的。唯一正确的是让这两步“要么一起成要么一起败”。这就是put the offset and the result in the same transaction的含义。Kafka给的一个官方示例是“consumer-producer”的经典结构用事务生产者将参与业务写入和位移提交用ack机制控制读取进度。这个写法有个天然的约束消费者的消费线程和处理线程必须解耦。你不能在poll的循环里同步等待事务提交因为Kafka消费者是单线程模型commitTransaction会阻塞poll导致consumer pull不到新消息、心跳超时等问题。正确做法是使用pause/resume控制消费节奏或者使用稳定的事务线程池专用提交。4.2 Kafka Streams 的 EOS 是怎么借力事务的如果你不想自己从零搭这套读-处理-写循环Kafka Streams是一个已经替你实现了EOS的选择。它的核心组件是任务与状态存储每个流任务有自己专属的分区集合和状态存储。在处理过程中状态更新会先以changelog消息写入内部topic再把结果发送到下游topic最后把该任务的一个或多个消费位移作为一个事务提交。Kafka Streams的EOS有一个参数叫processing.guarantee可选值为at_least_once和exactly_once_v2旧版本exactly_once在2.5之后被废弃。exactly_once_v2在设计上和v1有很大差异它针对每个task调用initTransactions而不是针对整个instance减少了事务协调器的负载同时解决了v1期间task迁移的僵尸实例问题。但用Kafka Streams的EOS不代表万事大吉它依然有两个红线第一Streams应用对外部数据库的写入不在事务保护范围内你仍然要自己处理外部系统幂等第二开启EOS后吞吐量下降因为每个任务都要经过事务提交和协调在数据量极大的场景下性能瓶颈明显。后来根据实际压测不少团队最终选择at_least_once加上外部幂等键去重性价比反而更高。4.3 离线/外部系统的最终一致方案对于外部系统的写入Kafka能给的只是“尽可能帮助简化幂等”而不是“保证外部系统幂等”。你可以在消息体里塞业务唯一键外部系统用这个键做去重也可以在外部系统里做状态机用消息幂等键查询表如果已处理则跳过。这是经典的“幂等消费”套路。有一个常被忽略的细节如果你的外部数据库是MySQL而且你正好在用InnoDB那么用主键冲突或唯一索引来去重是最高效的。Kafka的消息key可以设计成业务幂等键比如订单号写入MySQL时用INSERT IGNORE或ON DUPLICATE KEY UPDATE。这样即便Kafka重复消费数据库层面也不会产生重复数据。这个方案的优点是彻底的“开箱即用”缺点就是要自己处理消息key和业务键之间的映射对于分表分库的场景稍显麻烦。5. 配置参数逐项拆解与实际踩坑从理论落到实践需要把配套的参数一次说清。很多重复消息、脏读、事务超时的问题追根究底是配置没有成套使用。Kafka的配置项之间有很强的耦合性单改其中一个往往会引发别的连锁反应。5.1 必须成套看的核心参数先列一套典型的“正好一次”生产配置再逐个解释为什么这些参数要这样配。参数推荐值作用enable.idempotencetrue开启生产者幂等acksall配合幂等要求所有副本确认max.in.flight.requests.per.connection5默认保证同一分区消息顺序transactional.id唯一字符串启用事务并绑定协调器isolation.levelread_committed消费者只读已提交事务read_committed服务端过滤事务消息需要同时配置delivery.timeout.ms120000以上给事务重试留出足够时间transaction.timeout.ms默认60000未决事务自动中止阈值重点讲几个容易被忽略的。enable.idempotencetrue并不是单独生效的它要求acks必须是all。因为幂等机制依赖broker保存最大序列号这个保存动作必须在所有ISR副本都确认后才能稳定若acks不是all序列号记录可能丢失幂等等式也就无法成立。开启幂等后max.in.flight.requests.per.connection默认会被调整为5因为Kafka需要保证同一分区内消息按序到达如果允许任意乱序的in-flight请求幂等判断会完全失效。还有一个很容易误解的delivery.timeout.ms。它表示一条消息从发送开始到收到最终确认成功或失败的最长等待时间。在事务场景里如果你设置的delivery.timeout.ms小于事务完成所需总时长生产者会在事务还没提交前就判定消息发送失败后续commitTransaction就会抛异常。常见的错误是把retries设得很小又把delivery.timeout设得很短导致生产环境偶发事务提交失败的假象。5.2 事务超时那点事max.block.ms、transaction.timeout.ms事务相关的超时参数有两组经常被混为一谈。第一组是max.block.ms它控制生产者在send时的阻塞上限比如缓冲池满了、元数据拉取慢send会阻塞等待第二组是transaction.timeout.ms它控制事务从begin到commit的最长存活时间。两者互相独立但都有各自踩坑的场景。max.block.ms设置过短的典型现象是send方法频繁抛出TimeoutException。这个异常并不代表消息发送失败而是说阻塞时间到上限队列里积压了很多待发送消息。在事务场景里如果你在事务中连续发送大量消息缓冲池很快会被写满max.block.ms设短的话会直接导致事务被中断。transaction.timeout.ms则是你事务臃肿的元凶。默认60秒如果你的业务逻辑里除了写Kafka还包含外部系统调用比如等MySQL返回、调第三方接口整个事务周期可能超过60秒此时coordinator会自动abort。你会看到类似Aborting transaction with produced records because it has been more than 60000 ms since the transaction began的报错。排查这种问题要么把事务做小、把外部调用挪到事务外面要么单独加大transaction.timeout.ms。5.3 排查链路重复消息、脏读和延迟上涨讲三个生产环境里真实遇到过的故障帮大家建立排查直觉。第一个是重复消息。现象消费者处理了业务但日志里能看到同一条消息被处理了两次。排查顺序先看生产者配没配enable.idempotence确认之后再看是否用同一个transactional.id启用了多个实例如果多个实例互相抢写序列号会乱幂等记录也可能被覆盖。还有一种隐蔽情况老版本Kafka在有非幂等生产者的混布集群里事务消息和普通消息交错服务端的去重逻辑是隔离的重复只在读取端可见。这种时候要么升级客户端要么在消费端加一层业务幂等键兜底。第二个是脏读。现象isolation.level没改还是默认的read_uncommitted消费端读到了一些事务中写着但还没提交的数据。虽然是“脏读”但要说明的是Kafka这里的脏读和数据库里的脏读不完全一样——它不是读到被修改后未提交的值而是读到了“事务写入但未提交的消息”。一旦后续事务abort这些消息会以aborted marker标记但read_uncommitted消费者不会跳过它们。所以如果你的业务依赖Kafka事务消费端isolation.level必须明确设置为read_committed。第三个是延迟上涨。现象开启事务后消费组延迟从几十毫秒涨到几秒甚至几十秒。这个多见于读-处理-写模式。原因有两个一个是事务提交本身有额外RTT每个事务都需要和coordinator交互二是很多人在消费者线程内同步执行producer事务poll被阻塞整个消费管线停顿。解决办法是把生产者事务和消费逻辑解耦或者改用无界并发的事务线程池避免单个事务阻塞消费循环。6. EOS不是银弹成本模型与适用场景的边界写到这里该回答一个很多架构师都在纠结的问题既然Kafka能保证exactly once为什么不是所有项目都开答案很简单EOS有真实的成本而且它只覆盖Kafka内部消息流的原子性。不是所有业务都配得上这个成本。6.1 性能开销吞吐量测试的几组数字我在内部做过一组不严谨但很有参考价值的对比测试。单机空跑无事务的普通生产者写单分区吞吐约120万条/秒。开启幂等enable.idempotencetrue、acksall吞吐掉到约80万条/秒降幅两到三成。再开启事务每条消息走一次事务提交每条一个独立事务吞吐直接掉到20万条/秒以下降幅非常明显。如果你的场景是批量事务——把上千条消息放进一个事务里统一提交吞吐损耗会小得多。因为事务的固定开销在协调器交互上消息体本身的顺序写入依然高效。所以结论是事务“大而少”比“小而多”更划算把单个事务的规模做大才能摊薄协调成本。6.2 什么场景该用、什么场景不该用适合EOS的典型场景金融交易扣款、转账、余额变更重复执行会直接造成资损。对账系统端到端校验数据一致性要求中间过程不能有脏读。数据管道Kafka内部topic到topic的转换Kafka Streams EOS天然契合。不适合EOS的典型场景日志采集、埋点数据、监控指标这类数据天然带时间戳和唯一ID重复几跳无伤大雅用幂等生产者消费端去重就够了开启事务纯属浪费吞吐。大流量峰值系统直播弹幕、股票行情推送延迟和吞吐优先EOS的协调开销会让链路变重。已经有外部系统做幂等的架构你已经在MySQL里建了唯一索引、在Redis里做了状态标记再把Kafka事务开上属于多此一举成本加倍。6.3 我的最终推荐分级策略经过这些年的实践我对“Kafka不重不漏”的方案选择有了一个比较务实的梯队第一梯队也是所有项目都应该做的开启幂等生产者和acksall。这是最低成本、最高收益的改动几乎不牺牲性能却能消除重试导致的重复写入。第二梯队消费端使用at least once 业务幂等键去重。用消息key或业务唯一ID在数据库里做唯一约束/去重表兜住链路中所有可能导致重复的环节。这是兼容性最好的方案不依赖Kafka新特性也很容易在RocketMQ、RabbitMQ等队列间迁移。第三梯队Kafka内部纯消息流场景直接用Kafka Streams的exactly_once_v2省心省力。第四梯队跨系统强一致场景如果业务结果必须写入外部存储且无法接受最终一致那要评估的不只是Kafka而是整条链路的分布式事务。Kafka事务在这个场景只能帮你把“消费位移”这块收口外部系统的状态变更还是要靠你自己用幂等、状态机、对账去保证。Kafka不重不漏的答案从来不是某一个开关而是幂等生产者、事务API、消费隔离、外部系统幂等这四个模块的组合拳。把这些边界想清楚你离“生产环境里不再因为重复消息被业务方半夜叫醒”这个目标就真的很近了。
阅读完成 · 觉得有帮助？
