kafka消息送达语义说明 0.11 版本之前保证的语义是：至少一次 至少一次的解释 可以做到消息不丢失--> 可以做到发送成功的消息一定可以被消费到。 不能做到消息不重复。 ## 发送成功的消息，
0.11 版本之前保证的语义是：至少一次
至少一次的解释
可以做到消息不丢失--> 可以做到发送成功的消息一定可以被消费到。
不能做到消息不重复。
## 发送成功的消息，表示业务逻辑认为此消息已发送成功，即send方法已执行完成。
丢消息场景
异步发送端：
a：send之后，等待发送的时候down（消息在缓冲区中），导致消息丢失。
b：send时，缓冲区已满，导致消息丢弃。
同步（异步）发送端：
a：有限的重试
服务端Leader down时，因为有与zk的超时timeout，导致在timeout之后才会进行切换， 如果重试次数 * 重试间隔 < zk session timeout + 切换耗时，则消息会丢失。
b：ack ！= -1
ack = 0 的场景，不需服务端确认，发送后，Leader down，导致消息丢失。
ack = 1 的场景，只需要Leader确认，Leader收到消息后，未同步到Replica之前，Leader down，导致消息丢失。
服务端：
a：min.insync.replicas < 2 且 unclean.leader.election.enable = true
min.insync.replicas < 2 的场景下，如果副本均落后Leader，在Leader down时，根据脏选开关，会选择落后的副本作为新的Leader，则落后的数据会丢失。
消费端：
a：自动offset提交
消息处理失败，但是offset也提交了，对业务来说消息丢失。
b：先手动提交offset，后处理消息
先提交offset，后处理消息，但是处理逻辑失败，对业务来说消息丢失。
不丢失数据的方法
发送端：
a：同步发送
b：retries = Long.MAX_VALUE
c：acks = -1
服务端：
a：replication factor = 3
b：min.insync.replicas = 2
c：unclean.leader.election.enable = false
消费端：
a：auto.commit.enable = false
b：仅当消息处理成功之后再提交offset
消息重复的场景
发送端：
a：发送端重试
发送端发送消息后，服务端实际已接收，但是客户端因为网络或其他原因未收到确认响应， 再进行消息重试发送，导致消息重复。
消费端：
a：auto.commit.enable = true
消息处理后，为进行自动offset提交之前，consumer down，恢复后从上个提交点开始消费导致消息处理重复。
消息不重复的方法
发送端 --> 做不到。
消费端 --> auto.commit.enable = false，并且等消息处理完成再提交offset。
0.11之后版本保证的语义是：恰好一次
恰好一次的解释
可以做到消息不丢失 --> 可以做到发送成功的消息一定可以被消费到。
可以做到消息不重复。
发送端做不到消息不重复的解决办法
方法：
给Producer编号，且给每条消息编号，服务端保存Producer编号与此Producer最后一条消息的映射，服务端校验当前收到的消息是否与保存的最后一条消息编号相同，如果相同则拒绝。
配置：
enable.idempotence = true -->
retries = Integer.MAX_VALUE
max.in.flight.requests.per.connection = 1
acks = -1
过程：
a：Producer发送InitProducerIdRequest获取ProducerIdAndEpoch producerId 是Broker从zk分段获取后递增分配的，保证唯一，epoch为事务所用，暂时不提。
b：Producer发送消息（消息中加入ProducerIdAndEpoch，消息序列号（0））。
c：服务端收到消息，内存中维护 ProducerIdAndEpoch与此消息的映射，数据落盘， 同步到副本。
d：Producer发送消息（消息中加入ProducerIdAndEpoch，消息序列号（1））。
e：服务端收到消息，判断内存中的映射是否存在，如不存在，更新映射，数据落盘， 同步到副本，如存在，返回消息重复异常。
判断是否重复的条件：
isFromClient == true &&
batch.producerIdAndEpoch == producerIdAndEpoch &&
batch.baseSequence == firstSeq &&
batch.lastSequence == lastSeq
几个问题：
a：重试的场景中，ProducerIdAndEpoch映射的会不会不是上次重试的消息。
max.in.flight.requests.per.connection代表了同时只能有一条（批）消
息在发送，retries 保证了必须发送成功才会进行下一条的发送，所以不会有映射成其他消息 的情况。
b：有了映射缓存，其他副本没成功复制怎么办 acks = -1 保证了其他副本必须复制成功。
c：服务端启动或新选举为Leader的时候，缓存内容为空怎么判断重复 会先构造缓存内容再提供服务 初始化时加载.snapshot文件，append消息时更新缓存，退出时，写入.snapshot文件。
事务消息
事务消息语义
保证发送消息不重复。
保证多条消息（发往不同服务端）发送的原子性，即同时提交或同时回滚。
保证 “消费-处理-发送”逻辑的原子性，及要么全部成功，要么整体回滚。
整体原理
Producer新增 transactionalId、producerId、epoch标识，可标记事务。
新增transaction coordinator，协调处理事务的开启、提交、终止，记录事务日志 ；
transaction coordinator依附在Broker上，基于Broker提供服务；
记录事务日志是一个内部可根据Key压缩的Topic，可复制的，当 transaction coordinator down，有其他的接管继续事务处理。
服务端只要接收到消息，就写入文件，不管后续是否是提交或者终止。
新增control类型消息，标记事务是提交还是终止（全部消息以transactionalId为标识）。
消费端使用缓存堆积消息，直到看到control类型消息，返回给应用或丢弃。
多条消息发送的原子性
发送消息过程
a：初始化事务
发送FindCoordinatorRequest，查找transaction coordinator。
发送InitProducerIdRequest到transaction coordinator，获取ProducerIdAndEpoch。
transaction coordinator 新增消息 transactionalId --> producerId 到事务日志。
transaction coordinator 恢复或终止此 transactionalId 之前的事务。
b：开启事务
c：生产消息（可多次）
发送AddPartitionsToTxnRequest到transaction coordinator，transaction coordinator增加 BEGIN的事务日志，日志包含此partition 发送消息到实际的Broker，Broker接收消息写入文件并复制到副本
d：提交或终止事务
Producer 发送 EndTxnRequest 到 transaction coordinator transaction coordinator 写入PREPARE_COMMIT或 PREPARE_ABORT到事务日志 transaction coordinator 依次向全部partition发送 COMMIT/ ABORT的control消息 transaction coordinator 写入 COMMIT/ ABORT到事务日志 Broker接收消息写入文件并复制到副本
0