Kafka知识体系思维导图模板_ProcessOn思维导图、流程图

事务

http://matt33.com/2018/11/04/kafka-transaction

Kafka 事务性最开始的出发点是为了在 Kafka Streams 中实现 Exactly-Once 语义的数据处理，这个问题提出之后，在真正的方案讨论阶段，社区又挖掘了更多的应用场景，也为了尽可能覆盖更多的应用场景，在真正的实现中，在很多地方做了相应的 tradeoffs

txn.id 可以跟内部的 PID 1:1 分配，它们不同的是 txn.id 是用户提供的，而 PID 是 Producer 内部自动生成的（并且故障恢复后这个 PID 会变化），有了 txn.id 这个机制，就可以实现多 partition、跨会话的 EOS 语义

事务隔离级别（isolation.level）

read_uncommitted(默认)

消费者可以消费到 HW (High Watermark) 位置

read_committed

消费者忽略事务未提交的消息，即只能消费到 LSO (LastStableOffset) 的位置

LastStableOffset

该值会影响消息堆积量 Kafka Lag的计算，消息堆积可以通过自带工具查看 kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --group test2_consumer_group

1. 在隔离级别为 read_uncommited 的情况下，消息堆积量为 HW - ConsumerOffset

2. 在隔离级别为 read_commited 的情况下，消息堆积量为 LSO - ConsumerOffset

LSO存在一个问题，即当前若有一个 long transaction，比如其 first offset 是 1000，另外有几个已经完成的小事务操作，比如：txn1（offset：1100~1200）、txn2（offset：1400~1500），假设此时的 LSO 是 1000，也就是说这个 long transaction 还没有完成，那么已经完成的 txn1、txn2 也会对 consumer 不可见（假设都是 commit 操作），此时受 long transaction 的影响可能会导致数据有延迟。在实际的生产场景中，尽量避免 long transaction 这种操作

关于事务的部分思考

Consumer 如何过滤 abort 的事务数据

Broker 会追踪每个 Partition 涉及到的 abort transactions，Partition 的每个 log segment 都会有一个单独后缀为 .txnindex 的文件（append-only file）来存储 abort transaction 信息，因为 abort transaction 并不是很多，所以这个开销是可以可以接受的，之所以要持久化到磁盘，主要是为了故障后快速恢复，要不然 Broker 需要把这个 Partition 的所有数据都读一遍，才能知道哪些事务是 abort 的，这样的话开销太大（如果这个 Partition 没有事务操作，就不会生成这个文件）

有了这个设计，Consumer 在拉取数据时，Broker 会把这批数据涉及到的所有 abort transaction 信息都返回给 Consumer，Server 端会根据拉取的 offset 范围与 abort transaction 的 offset 做对比，返回涉及到的 abort transaction 集合

PID Snapshot 是做什么的？用来解决什么问题？

对于每个 Topic-Partition，Broker 都会在内存中维护其 PID 与 sequence number（最后成功写入的 msg 的 sequence number）的对应关系。

Broker 重启时，如果想恢复上面的状态信息，那么它读取所有的 log 文件。相比于之下，定期对这个 state 信息做 checkpoint（Snapshot），明显收益是非常大的，此时如果 Broker 重启，只需要读取最近一个 Snapshot 文件，之后的数据再从 log 文件中恢复即可

若 txn.id 长期不使用，server 端怎么处理

producer 端默认事务超时时间为60s transaction.timeout.ms 。Producer 设置超时时间不能超过 Server，否则会抛出异常

broker端默认事务超时时间为 15min, transaction.max.timeout.ms

对于 txn.id，我们知道 TransactionCoordinator 会缓存 txn.id 的相关信息，如果没有超时机制，这个 meta 大小是无法预估的，Server 端提供了一个 transaction.id.expiration.ms 参数来配置这个超时时间（默认是 7 天），如果超过这个时间没有任何事务相关的请求发送过来，那么 TransactionCoordinator 将会使这个 txn.id 过期

Producer 幂等性

幂等性实现是事务性实现的基础，幂等性提供了单会话单 Partition Exactly-Once 语义的实现，正是因为 Idempotent Producer 不提供跨多个 Partition 和跨会话场景下的保证

只能保证 Producer 在单个会话内不丟不重，如果 Producer 出现意外挂掉再重启是无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不丢不重）

幂等性不能跨多个 Topic-Partition，只能保证单个 partition 内的幂等性，当涉及多个 Topic-Partition 时，这中间的状态并没有同步。

幂等性用来解决什么问题

在 0.11.0 之前，Kafka 通过 Producer 端和 Server 端的相关配置可以做到数据不丢，也就是 at least once，但是在一些网络异常情况下，可能会导致数据重复

PID + Sequence Number

PID 用于标识每一个 producer

sequence numbers，client 发送的每条消息都会带相应的 sequence number，Server 端就是根据这个值来判断数据是否重复。sequence number 将会从 0 开始自增，每个 Topic-Partition 都会有一个独立的 sequence number

kafka的事务保证

跨会话的幂等性写入：即使中间故障，恢复后依然可以保持幂等性；

跨会话的事务恢复：如果一个应用实例挂了，启动的下一个实例依然可以保证上一个事务完成（commit 或者 abort）；

跨多个 Topic-Partition 的幂等性写入，Kafka 可以保证跨多个 Topic-Partition 的数据要么全部写入成功，要么全部失败，不会出现中间状态。

TransactionCoordinator 脑裂问题(brain split)

发生场景

TransactionCoordinator 在遇到上 long FGC 时，可能会导致脑裂问题，FGC 时会 stop-the-world，这时候可能会与 zk 连接超时导致临时节点消失进而触发 leader 选举，如果 __transaction_state 发生了 leader 选举，TransactionCoordinator 就会切换，如果此时旧的 TransactionCoordinator FGC 完成，在还没来得及同步到最细 meta 之前，会有一个短暂的时刻，对于一个 txn.id 而言就是这个时刻可能出现了两个 TransactionCoordinator

解决方式（Fencing）

通过 CoordinatorEpoch 来判断，每个 TransactionCoordinator 都有其 CoordinatorEpoch 值，这个值就是对应 __transaction_state Partition 的 Epoch 值（每当 leader 切换一次，该值就会自增 1）

zk, raft 的解决方式

与 HDFS 不同(Fencing)

HDFS 选举得到NN后会在ZK创建一个持久节点，用于记录当前NN的地址信息。当NN会话状态发生正常切换时，该持久节点也会被删除。而异常状态下该节点不会被删除，故NN选举成功后可以检查是否存在该持久节点，存在则说明出现了脑裂的情况，直接 Kill 旧的 NN 进程或者强制切换状态

疑问

为什么kafka不支持读写分离

1. 数据一致性，leader同步到ISR中的副本存在一定时间差，容易造成数据不一致

2. 负载不均衡

延迟消息

Timingwheel

异常

CommitFailedException

情况1：消费处理业务逻辑时间超出session.timeout.ms或者max.poll.interval.ms（看版本）超时后会导致当前consumer已崩溃或宕机，导致被group踢出组后重新发生rebalance,此时consumer再提交消息就会发现该consumer已经不在group组内

情况2：standalone consumer与consumer group冲突时：这里所说的standalone consumer指的是使用KafkaConsumer.assign()而非subscribe()的消费者。当用户系统中同时出现了standalone consumer和consumer group，并且它们的group id相同时，此时standalone consumer手动提交位移时就会立刻抛出此异常。这是因为group coordinator无法识别这个具有相同group id的consumer，从而向它返回“你不是一个合法的组成员”错误

版本历史

0.7

基本消息队列，无副本机制

0.8

引入副本机制，真正意义上完备的分布式高可靠消息队列

0.9

增加基础安全认证及权限功能

使用Java重写新版本consumer API

引入Kafka Connect组件用于实现高性能的数据抽取

新的Comsumer API

Kafka可以自行维护Offset、消费者的Position。也可以开发者自己来维护Offset，实现相关的业务需求。

新的ComsumerAPI不再有high-level、low-level之分了,而是自己维护offset。这样做的好处是避免应用出现异常时，数据未消费成功，但Position已经提交，导致消息未消费的情况发生。通过查看API

自行控制Consumer消费消息的位置

可以使用外部存储记录Offset，如数据库之类的

可以使用多线程进行消费

0.10

引入Kafka Streams，正式成为流式处理框架

内置机架感知以便隔离副本，提高可用性

消息格式增加时间戳

引入max.poll.records参数，允许开发者控制返回消息的条数

0.11

实现幂等性producer

支持事务

重构消息格式

新的分配算法 Sticky

重构controller

老版本controller在执行多步操作时，若其中一步出错，则无法回滚之前的操作

多线程同时访问controller上下文，重构后采用单线程+基于事件队列的方式

1.x (2017.11)

改进Streams Api

改进Connect的度量指标

支持Java9，实现更快的TLS和CRC32, 加快加密速度降低计算开销

调整了 SASL 认证模块的错误处理逻辑，原先的认证错误信息现在被清晰地记录到日志当中

更好地支持磁盘容错，更优雅地处理磁盘错误，单个 JBOD 上的磁盘错误不会导致整个集群崩溃。

0.11.0 版本中引入的幂等性生产者需要将max.in.flight.requests.per.connection 参数设置为 1，这对吞吐量造成了一定的限制。而在 1.0.0 版本里，这个参数最大可以被设置为 5，极大提升了吞吐量范围。

消费顺序

Kafka仅提供单个分区内的消费顺序，而不会维护全局的消费顺序

若要实现全局的消费顺序就只能通过让每个consumer group下只包含一个consumer实例来间接实现

Broker

zookeeper路径

/brokers

/ids

/<broker.id>保存成员节点信息

version

host

port

jmx_port

timestamp

endpoints

指明传输协议类型、主机名及端口号，如PLAINTEXT://host1:9092

rack

机架信息，若设置了该信息则Kafka在分配副本时会考虑把某个分区的多个副本分配到多个机架上，以保证高可用

listener_security_protocol_map

broker与外界通信的安全协议类型

/topics

/seqid

/controller

/admin

/delete_topics

/reassign_partitions

/preferred_replica_election

/isr_change_notification

/config

/topics

/changes

/clients

/users

/brokers

/cluster

保存j集群的简要信息

/controller_epoch

保存controller组件版本号，Kafka使用该版本号来隔离无效的controller请求

副本同步

术语及概念

AR(Assigned Replicas)

分区中的所有副本统称为 AR

ISR(In-Sync Replicas)

如何判定ISR

0.9.0版本之前

replica.lag.max.messages 用于控制follower落后leader副本的消息数

瞬时高峰流量可能会大于该消息数，故会导致 follower被不断的踢出加入，再踢出再加入

replica.lag.time.max.ms 表示若follower副本无法在该时间内向leader请求数据，则会被踢出ISR

0.9.0版本之后

统一使用参数 replica.lag.time.max.ms，默认10s

follower与leader不同步的原因

1. 请求速度追不上

2. 进程卡住，如频繁GC或程序bug

3. 创建新副本

OSR (Out-of-Sync Replicas ）

与 leader 副本同步滞后过多的副本

起始位移 (base offset / LogStartOffset)

该副本第一条消息的offset

logStartOffset 受日志清除策略影响，注意 LogStartOffset 不可以缩写为 LSO，因为在 Kafka 中，LSO 特指 LogStableOffset（与事务实现息息相关）

高水位值 (High Watermark, HW)

表示该副本最新一条已提交消息的下一条位移，消费端只能拉取到此offset之前的消息

更新机制

leader会尝试更新HW的情况

1. follower成为leader时

2. broker出现崩溃导致副本被踢出ISR时

3. producer向leader写入消息时

4. leader处理follower的FETCH请求时：leader处理follower的FETCH请求时，首先会从底层的log读取数据，之后再尝试更新分区的HW

leader更新HW

leader broker尝试确定分区HW时，会选出所有满足条件的副本比较他们的LEO以及自己的LEO，并选择最小的LEO最为HW

follower更新HW

follower更新HW发生在其更新LEO之后，一旦follower向log写完数据，它就会尝试更新HW。具体算法是比较当前LEO与FETCH响应中leader的HW值，取两者的最小值作为HW

完整流程

情况1：leader副本写入消息后， follower副本发送FETCH请求

1. leader接收到生产者消息后将其写入到底层日志，同时更新自身LEO

2. 尝试更新leader的HW：follower尚未发起FETCH请求，故leader保存的remote LEO=0，将remote LEO与leader LEO对比取最小值即 HW = 0，与原HW相同，故不更新分区HW

3. leader 接收到follower的FETCH请求，开始读取底层log数据，并根据FETCH请求中的fetch offset=0 更新remote LEO

4. leader再次尝试更新HW，leader LEO=1，remote LEO =0，故分区HW=0，此时consumer无可消费的消息

5. leader将数据及当前HW返回给follower

6. follower接收到FETCH响应后将消息写入log日志并更新自身 LEO = 1

7. follower根据响应返回的HW与本地LEO做对比，取其中最小的值来更新HW，故HW=0。此时consumer仍无可消费的消息

8. follower发起第二轮FETCH请求，同时带上参数fetch offset=1(自身的LEO=1)发送给leader

9. leader 接收到请求后读取底层log数据，并更新自身的remote LEO=1

10. leader 尝试更新分区HW，对比自身LEO=1 与 remote LEO=1 ，取最小值. 即HW=1, 原HW=0，此时HW被更新

11. leader将数据（空数据）及当前的HW返回给follower

12. follower接收到响应后将消息写入log，因为是空数据所有无消息可写，LEO仍然为1

13. follower更新自身HW，对比本地LEO=1与fetch请求返回的HW=1，获取最小值即1，故follower 的HW=1。此时消息已经被成功拷贝到leader及follower的log中且分区HW=1，所以consumer可消费offset=0的消息

情况2：FETCH请求保存在 purgatory中时生产者发来消息

原理与情况1基本类似

缺陷

备份数据丢失

前提是min.insync.repicas=1

场景

1. 有两个副本A,B，A为leader。生产者发送两条消息给A，A写入成功并成功返回

2. 此时leader及follower都已写入两条消息，且分区HW已更新为2，但follower的HW未被更新，需等到下一轮FETCH请求方可更新。此时 A（hw=leo=2）,B(hw=1,leo=2)

3. 此时B宕机，则重启B后自动将LEO调整为之前的HW，故B会做日志截取，将offset=1的消息从log中删除，并调整LEO=1. 即 B （HW=LEO=1）

4. B重启后需要向A发起FETCH请求，但此时A宕机，B成为新的leader。而当A重启后又向B发起FETCH请求，对比自身LEO与leader的HW，发现应该取leader的HW=1，故A做日志截取，将offset=1的消息删除。此时 offset=1的消息在集群所有副本中被永久删除

总结

延迟一轮FETCH请求更新HW值的设计使得follower HW值是异步延迟更新的，若这个过程中leader发生变更，那么成为新leader的follower的HW值就有可能是过期的

备份数据不一致

前提是min.insync.repicas=1

场景

1. 有两个副本A,B，A为leader。生产者发送两条消息给A，A写入成功并成功返回

2. 此时leader及follower都已写入两条消息，且分区HW已更新为2，但follower的HW未被更新，需等到下一轮FETCH请求方可更新。此时 A（hw=leo=2）,B(hw=1,leo=2)

3. A,B同时宕机，若B先重启，成为leader，分区HW=1。此时producer发送第三条消息给B，B写入成功也提交成功，因为只有B自己一个副本, 即 B（HW=LEO=2）

4. 此时A重启，执行FETCH请求，发现自身LEO=2，而leader的HW=2，两者相同无需更新。这就造成leader中offset=1的消息与follower中offset=1的消息不一致问题

LW ( Low Watermark)

AR 集合中最小的 logStartOffset 值

日志末端位移 (Log End Offset, LEO)

副本日志中下一条待写入消息的offset。每当leader接收到producer推送的消息，它会更新自己的LEO，通常是加1

每个副本都由LEO，不论leader与follower。leader除保存自己的LEO以外还保存这所有follower的LEO

同步流程

1. broker上的leader副本接收到消息后将自己的LEO值加一

2. follower副本各自发送请求给leader

3. leader分别将该消息推送给follower副本

4. follower副本接收到消息后各自将更新自己的LEO，加一

5. leader副本接收到其他follower副本的响应后，更新HW。也即代表该条消息可以被consumer消息

日志存储

每个topic都会有对应的子目录，目录名称为 <topic>_<分区号>

文件结构

日志段 .log

以该文件的第一条消息对应的offset来命名

log.segment.bytes 指定文件大小，默认1G.日志段被填满后，Kafka会自行切分，创建出新的日志段及索引文件

索引文件

分类

.index 位移索引文件

Kafka强制要求必须是8的整数倍，因为每个索引项占用8字节

.timeindex 时间戳索引文件(0.10.0引入)

Kafka强制要求必须是12的整数倍. 因为每个索引项占用12字节

log.index.size.max.bytes 索引文件大小，默认10M.

方便通过二分查找快速定位（O(LogN)）

日志清除策略

1. 基于时间清除

默认清除7天前的日志数据（包含日志段文件及两个索引文件）

log.retention.{hours|minutes|ms} 可配置时间间隔，ms优先级最高，minutes次之，hours最后

0.10版本之前使用当前时间与文件最近修改时间做比较，而该时间会时常变动。因此0.10开始取日志段首条消息的时间做对比

2. 基于文件大小清除

log.retention.bytes, 默认-1，对log大小不做限制

3. 定时清除

在 Kafka 的日志管理器中会有一个专门的日志删除任务来周期性地检测和删除不符合保留条件的日志分段文件，这个周期可以通过 broker 端参数 log.retention.check.interval.ms 来配置，默认值为 300000，即 5 分钟

LogSegment保留策略

基于时间的保留策略

基于日志大小的保留策略

基于日志起始偏移量的保留策略

基于 logStartOffset 来实现

日志压缩（compaction）

对于相同key不同value的消息，仅保留最新的一条，多余的会被清除

必须设置可以，否则无法执行该操作

使用场景

高可用日志化

数据库变更订阅

时间溯源