MQ - KAFKA 高级篇

kafak是一个分布式流处理平台,提供消息持久化,基于发布-订阅的方式的消息中间件，同时通过消费端配置相同的groupId支持点对点通信。

##适用场景：

1. 构造实时流数据管道,用于系统或应用之间可靠的消息传输.
2. 数据采集及处理,例如连接到一个数据库系统,捕捉表的变更内容.
3. 构建实时流式应用程序，对这些流数据进行转换或者影响,如:应用程序作为一个流处理器，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或多个topic中去，在输入输出流中进行有效的转换.
4. 应用耦合：多应用间通过消息队列对同一消息进行处理，避免调用接口失败导致整个过程失败；
5. 异步处理：多应用对消息队列中同一消息进行处理，应用间并发处理消息，相比串行处理，减少处理时间；
6. 限流削峰：广泛应用于秒杀或抢购活动中，避免流量过大导致应用系统挂掉的情况；
7. 消息驱动的系统：系统分为消息队列、消息生产者、消息消费者，生产者负责产生消息，消费者(可能有多个)负责对消息进行处理；

##特性：
8. 生产者/消费者支持多语言。
9. 支持分布式横向扩缩容。
10. 高性能(高吞吐量)。
11. 版本向下兼容。
12. 提供消息持久化。
13. 流处理。

##高性能实现：
14. 磁盘顺序读取和写入(接近内存随机读写的性能)。
15. nio和零拷贝。
16. 消息批处理。
17. 消息压缩

##多个producer和多个consumer group及zk示意图如下：
其中zookeeper是基于zab分布式协议实现的一个组件,主要实现对broker集群起协调作用,详细的后面文章再探讨.
注意图中的消费者和生产者,均是仅和分区的leader相连,所有的flower不参与外部交互,在内部作为leader的消费者去拉去消息进行追赶.
其中broker中还有几个概念

##问：kafka是如何保证它的高吞吐量？
1.消息生产者批量发送压缩消息
结合磁盘顺序写入，批量处理无疑是非常有必要（如果用的时候每发送一条消息都调用future.get等待，性能至少下降2个数量级）。写入的时候放到RecordAccumulator进行聚合，批量压缩，还有批量刷盘等…
producer批量并压缩消息–>broker直接落盘–>consumer批量获取消息并解压
批量和压缩可以大大降低网络开销和磁盘io开销,极大的提高吞吐量,且消息体越大,单批消息越多,效果越好.
broker端有个压缩格式的配置,默认跟从producer配置,若是指明了压缩格式后,则必须保证producer和broker一致,否则将会导致0拷贝失效,建议采用默认配置,既该配置会尊重producer的压缩格式.

简单来讲，整个生产者客户端（java版本）由两个线程协调运行，这两个线程分别为主线程和Sender线程（发送线程）。在主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器（RecordAccumulator，也称为消息收集器）中。Sender 线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。RecordAccumulator 主要用来缓存消息以便 Sender 线程可以批量发送，进而减少网络传输的资源消耗以提升性能。
2、pageCache的使用

kafka没有选择In-Process Cache的方式,而是在消息写入和读取的过程中充分的利用了操作系统的页缓存及磁盘预读取等特性。
pageCache避免在JVM内部缓存数据,当broker重启时,由于避免了缓存加载到jvm中的过程,大大加快broker的恢复速度,同时可避免不必要的GC,大大节约内存占用.磁盘预读取则有效的降低了磁盘io次数.kafka的消息读取和写入,时间复杂度为O(1)
3、零拷贝(Zero-Copy) （包括kafka收到消息写和kafka发出消息读）
由原来的四次拷贝转换为两次拷贝，这是其一；同时也减少了内核态和用户态的切换开销。
零拷贝是有硬件条件支持的，即DMA:DMA(Direct Memory Access，直接内存存取) 是所有现代计算机的重要特色，它允许不同速度的硬件装置直接沟通，而不需要依于CPU的大量中断负载.在现代计算机中,运算单元不再仅仅是cpu。网卡/磁盘等都可以认为是DMA设备,是一个半自治单元,比如网卡有它自己的运算单元(相当于特异化的cpu)和自己的缓存,网卡接收和发送数据时是不需要cpu的全程参与的,磁盘也是类似的.简单来讲就是dma设备就是cpu领导下的一个不太聪明的小弟,cpu负责指挥小弟去干活,但干活的过程中是不需要cpu参与的.nio和0拷贝都是为了解放cpu。
4、磁盘顺序读写（包括kafka收到消息写和kafka发出消息读）
kafka采用日志append的方式,一直在文件的末尾追加消息,既顺序写入,该方式比内存的随机写还要快一些，相当于按住磁头不动一直写，不需要多于的磁柱旋转时间和磁头寻址时间。还有一点是一个ProducerBatch是4KB，OS每次写是(8*512B)=4KB
5、Kafka二分查找定位数据
Kafka里面每一条消息，都有自己的offset（相对偏移量），存在物理磁盘上面，在position Position：物理位置（磁盘上面哪个地方）也就是说一条消息就有两个位置：offset：相对偏移量（相对位置）position：磁盘物理位置 稀疏索引： Kafka中采用了稀疏索引的方式读取索引，kafka每当写入了4k大小的日志（.log），就往index里写入一个记录索引。其中会采用二分查找

##问kafka什么情况下零拷贝失效？
1客户端的压缩格式和服务的压缩格式不一样
2由于消息格式的变动,若是集群版本和客户端版本不一致,有可能broker为了协议兼容,需要做消息格式转换,此转换会导致kafka的零拷贝失效,生产环境的表现就是kafka集群的性能大幅下降,尤其需要注意的是kafka的0.10.x和0.11.x,这两个版本把kakfa的消息格式分为了三个版本。
零拷贝可参考如下几篇优秀文章：

Linux I/O 原理和 Zero-copy 技术全面揭秘
张彦飞：图解Linux网络包接收过程
零壹技术栈：深入剖析Linux IO原理和几种零拷贝机制的实现

##问kafka的消息传递过程？
1、网络数据持久化到磁盘 (Producer 到 Broker) mmp
2、磁盘文件通过网络发送（Broker 到 Consumer） sendfile
从上面的分析可知,显然第一步,producer–>broker的过程可以利用到mmp内存映射
把磁盘->内核内存->应用内存,其中内核内存和应用内存映射后实际是同一块物理内存,则broker接收到数据后直接写入对外内存后,就相当于写入内核缓冲区,内核直接可以写入到磁盘中.而且mmap映射技术,修改内存就等价于修改了磁盘内容,是操作系统同步的.此处可以看到至少减少了两次内存复制,既jvm的堆外/堆内及应用内存和内核内存的复制.
而第二步消费者获取数据的过程,利用了linux的sendfile技术,直接把磁盘上的文件通过网络发送出去,没有多余的拷贝.kafka能实现上述操作其实还是得益于kafka的架构设计,broker不需要对数据做特别的修改,否则将会导致0拷贝失效.

##问说一下kafka中的ISR？
大部分分布式一致性协议都要满足一半以上的参与者投票来保证一致性,但kafka考虑到性能和一致性,做了折中,ISR，也即In-sync Replica。每个Partition的Leader都会维护这样一个动态列表，该列表中，包含了所有与之同步的Replica（包含Leader自己）。每次数据写入时，只有ISR中的所有Replica都复制完，Leader才会将其置为Commit，它才能被Consumer所消费。
这种方案，与同步复制非常接近。但不同的是，这个ISR是由Leader动态维护的。如果Follower不能紧“跟上”Leader，它将被Leader从ISR中移除，待它又重新“跟上”Leader后，会被Leader再次加加ISR中。每次改变ISR后，Leader都会将最新的ISR持久化到Zookeeper中。
是否同步的判断,低版本采用落后时间+落后消息数量来判断,高版本(0.9及以上)则仅通过落后的时间来判断.