深入理解Kafka：核心设计与实践原理  
https://book.douban.com/subject/30437872/

https://github.com/hiddenzzh/kafka_book_demo




1. 初识 Kafka
消息系统 分区顺序性保证和回溯消费 
存储系统 持久化和多副本
流式处理平台 流式处理类库 

leader 维护 ISR(in-sync Replica)，每个Partition都会有一个ISR，由leader动态维护
如果一个follower落后太多，从ISR中移除
当ISR中所有Replica都向Leader发送ACK时，leader才commit， HW机制 ，commit 的消息才对消费者可见

Producer, Broker, Consumer
通过 ZooKeeper 管理元数据、控制器的选举等操作 

主题(topic) 生产者发送消息到特定主题，消费者订阅主题并消费
分区(partition)
消息存储 一个可追加的日志文件
偏移量(offset) 消息在分区中的唯一标识，保证分区内的有序性
分区 多副本机制(replica)，副本之间是一主多从关系，leader 负责写请求，follower同步消息



2. 生产者

发送消息三种模式
发后即忘（fire-and-forget） 同步（sync） 异步（async）
fire-and-forget 直接调用 send()
sync   调用 send 之后调用 get() 等待
async: 传入回调 callback

自定义分区器
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());




3. 消费者

消费者和消费者组
每个消费者都有一个对应的消费者组(逻辑概念)，消息发布到topic 后，只会投递给订阅它的每个消费组中的一个消费者。 
消费者组是Kafka实现单播和广播两种消息模型的手段


配置消费者客户端参数及创建相应的消费者实例
订阅主题
拉取消息并消费
提交消费位移
关闭消费者实例


offset 
自动位移提交 enable.auto.commit=true
设置为 false 可以手动提交
commitSync
commitAsync
seek 方法可以从特定位置读取消息



4. 主题和分区


主题管理
kafka-topics.sh --zookeeper localhost:2181/kafka  --create --topic topic001 --partitions 6 --replication-factor 2

分区管理
leader 副本对外提供读写服务，flollwer 副本只负责数据同步

选择合适分区数
kafka-producer-perf-test.sh
kafka-consumer-perf-test.sh
分区数超过某个阈值之后，相应的吞吐量也会下降



5 消息存储 
文件目录
日志格式演变
v1 比 v0 多了时间戳

compression.type 压缩方式或者不压缩

日志索引
偏移量索引文件用来建立消息偏移量（offset）到物理地址之间的映射关系，方便快速定位消息所在的物理文件位置；
时间戳索引文件则根据指定的时间戳（timestamp）来查找对应的偏移量信息。

日志清理
两种日志清理策略

日志除(log retention): 删除不符合条件的日志分段
日志压缩(log compaction): 针对每个消息的 key 进行整合，对于相同 key 的不同 value 值，只保留最后一个版本

磁盘存储
顺序写磁盘速度很快，甚至比 随机 写内存快
使用 顺序追加、页缓存、零拷贝等技术提升性能

零拷贝(zero-copy)
将数据直接从磁盘文件复制到网卡设备中，减少内核和用户模式之间的上下文切换 和 数据拷贝
依赖 linux sendfile() ，DMA(Direct Memroy Access) 


6.服务端

协议设计
自定义一组基于 tcp 的二进制协议，遵守协议就可以发消息给 kafka。
每种协议类型由对应的请求和响应，Request 包含请求头和请求体。

基于时间轮实现用于延时功能的定时器(SystemTimer)，插入和删除O(1)。 
Kafka 中的时间轮(TimingWheel)是一个存储定时任务的环形队列，
底层采用数组实现，数组中的每个元素可以存放一个 定时任务列表(TimerTaskList)，
TimerTaskList是一个环形的双向链表，链表每一项都是定时任务项(TimerTaskEntry)， 其中封装了定时任务（TimerTask)。

一个或者多个 broker，其中一个 broker 会被选举为控制器(kafka controller)，负责管理整个集群中所有分区和副本的状态。


7 客户端
partition.assignment.strategy 
消费者和订阅主题之间的分区分配策略:

RangeAssignor
RoundRobinAssignor
StickyAssignor  分区尽可能均匀；尽可能和上次分配一致

消费者协调器和组协调器
GroupCoordinator  服务端用于管理消费组的组件
消费者客户端   ConsumerCoordinator 组件负责和 GroupCoordinator 交互
如果消费者发生变化触发再均衡操作

 __consumer_offsets 
位移保存到 内部主题 __consumer_offsets 中
使用 kafka-console-consumer.sh 查看 __consumer_offsets 中的内容

事务
消息传输保障

at most once(至多一次)
at least once(最少一次)
exactly once(恰好一次)
0.11.0.0 版本引入幂等和事务这两个特性，实现 EOS(exactly once semantics)

幂等
客户端参数 properties.put("enable.idempotence", true); 

为了实现生产者幂等，引入 producer id 和序列号 sequence number 
broker 在内存中维护一个序列号，
只有消息序列号的值(SN_new)比 broker 中维护的 SN_old 大 1， broker 才会接受
(SN_new=SN_old+1)。

只能保证单个生产者会话 (session) 中单分区的幂等

事务可以保证多个分区写入操作的原子性
properties.put("transactional.id", "transactionId") 



8 可靠性探究
当 ISR 集合中的一个 follower 副本滞后 leader 副本的时间超过 replica.lag.time.max.ms  则同步失败

日志同步机制
PacificA 算法

副本数：一般设置为 3 
客户端 acks 设置
acks=-1 leader 副本在写入本地日志之后还要等待 ISR 中的 follower 副本全部同步完成才告知生产者成功提交
同步刷盘策略（一般应该由多副本保证）
broker 参数 log.flush.interval.messages 和 log.flush.interval.ms 调整同步刷盘策略 
影响性能

enable.auto.commit 自动位移提交 可能导致 重复消费和消息丢失的问题



9. Kafka 应用

命令行工具

kafka.consumer-groups.sh 查看或者变更消费组信息
kafka-consumer.groups.sh 重置消费者组内消费位移
kafka-delete-records.sh 删除指定位置前的消息

Kafka Connect


两个核心概念
Source 负责导入数据到 kafka
Sink 负责从 kafka 导出数据

connect-standalone.sh 独立模式运行 kafka connect。

Kafka Mirror Maker
集群之间同步数据

Kafka Streams
高吞吐、高可用、低延时 
流式处理 


10. Kafka 监控

监控维度：集群信息、broker 信息、主题信息和消费组信息 

监控数据来源
集群层面的指标可以用 JMX  

消费滞后 Lag

同步失效分区
同步失效副本，包含它的分区就是同步失效分区。

监控架构主要分为数据采集、数据存储、数据展示三个部分



11. 高级应用
过期时间（TTL)
TTL(Time To Live，过期时间)  在消费者客户端拦截器实现

延时队列

订单系统，下单之后30 分钟内进行支付，没有成功支付则关闭订单 


死信队列和重试队列
由于某些原因消息无法被正确投递，一般放到一个特殊的队列称为死信队列。 
后续通过消费这个死信队列中的内容来分析当时遇到的异常情况，改善和优化系统。

消息路由
默认用主题进行路由
可在消息headers中加入信息，消费端中使用拦截器挑选特定业务标识消息

消息轨迹
通过封装客户端，保证正常生产消费的同时添加相应的轨迹信息埋点逻辑

消息审计
消息生产、存储和消费的整个过程之间对消息数和延迟的审计，检测是否有消息丢失、是否数据重复、端到端延迟等

消息代理
Kafka Rest Proxy 
Rest 接口  实现发送消息、消费消息、查看集群状态、执行管理操作等功能

The Kafka REST Proxy provides a RESTful interface to a Kafka cluster. 
It makes it easy to produce and consume messages, view the state of the cluster, 
and perform administrative actions without using the native Kafka protocol or clients.

https://github.com/confluentinc/kafka-rest

https://github.com/confluentinc/kafka-rest/blob/master/kafka-rest/src/main/java/io/confluent/kafkarest/controllers/TopicManagerImpl.java

消息中间件选型
RabbitMQ/Kafka/RocketMQ

选型维度  功能维度 性能维度



功能维度
优先级队列
延时队列
重试队列
死信队列
消费模式: push/pull
广播消费: 点对点(p2p)，还是发布订阅(Pub/Sub)
回溯消费
消息堆积+持久化
消息轨迹
消息审计
消息过滤
多租户
多协议支持
跨语言支持
流量控制：调整发送速度。通常的流控方法有 stop-and-wait，滑动窗口、令牌桶等
消息顺序性
安全机制
消息幂等性。消息语义：至多一次；至少一次；精确一次
事务性消息：要么发送成功，要么失败

性能维度
可靠性和可用性。金融支付领域 RabbitMQ 多，日志处理、大数据 kafka 多。
运维管理。申请、审核、监控、告警、管理、容灾、部署等
社区和生态


12. Kafka 和 Spark 集成