Kafka

Kafka简述

Kafka is written in Scala and Java.
publish-subscribe based fault tolerant messaging system
for big data :
大数据的两个主要挑战：
1. 如何收集大量的数据
2. 分析收集到的大量数据
To overcome those challenges, you must need a messaging system.

Kafka为分布式应用提供高吞吐量的systems，Kafka是传统的消息中间件很好的替代

1. 高吞吐量
2. built-in partitioning
3. 复制
4. 内在高容错（inherent fault-tolerance）
5. 高可靠性（分布式，分区，复制，容错）
6. 可扩展性
7. 持久性，Kafka uses Distributed commit log which means messages persists on disk as fast as possible, hence it is durable..
8. 高性能

什么是消息中间件：

transferring data from one application to another,
这样application只要专注data，而不用考虑如何share it，或者和其他应用服务之间如何交互共享数据。
消峰
解耦
异步
Distributed messaging ：reliable message queuing可靠的消息队列

1. point-point：
   在point-point，消息存在在消息队列中，FIFO，一旦队列中消息被消费了，就会从队列中移除。典型为Order Processing System
2. publish-subscribe（pub-sub）：发布-订阅
   消息放在topic中，用户可以订阅1个或多个topic

为什么要使用消息中间件:

1. 解耦
2. 冗余:
   数据处理过程失败，除非数据持久化，否则会造成数据的丢失，消息队列把数据进行持久化直到他们完全被处理
   “插入-获取-删除”，把一个消息从队列中删除时，需要处理过程中明确指出该消息已被处理完毕,确保数据被安全的保存到直到使用完毕
   ？？？如何实现？？？？
3. 扩展性: 因为消息队列解耦了处理过程
4. 灵活性和峰值处理能力: ！！！重点
5. 可恢复性:
   一个消息队列的处理进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理， 不丢失？
6. 送达保证:
7. 顺序保证: 如何消息处理失败呢？
8. 缓冲
9. 异步通信

什么是Kafka？
    首先，Kafka是一个分布式pub-sub的消息系统，robust queue能够handle大量的数据
    Kafka is suitable for both offline and online message consumption
    Kafka的messages储存在硬盘上，并且在cluster中复制，防止数据丢失
    Kafka上层为Zookeeper 同步服务，对于实时数据流处理分析集成非常好（Storm ，Spark）
    Kafka是一个统一处理实时数据feeds的统一平台，Kafka支持低延迟的数据传送以及高容错的保证，有能力handle大量的consumer。
    Kafka非常快，把所有数据都保存在disk上，这实际上意味着所有写入都会转到OS（RAM）的页缓存中。 
    这使得将数据从页面缓存传输到网络套接字非常有效。

传统IO，数据先从磁盘复制到内核空间缓冲区，然后从内核空间缓存区复制到应用程序的地址空间。这里的内核缓冲区也就是页缓存--Page Cache，是虚拟空间
分离内核空间和用户空间，来改善IO效率，直接利用操作系统的Page来实现文件到无力内存的直接映射
好处：
Reliability
Scalability
Durability
Perfromance
常用场景：
Metrics

    Log Aggregation Solution
    
    Stream Processing

Kafka常用术语

1. Topics:类似queue，一组消息的集合,发送给特定的category，topic被分成若干个分区，最少为一个，不可变有序序列，每一个分区实现 一组大小相等的段文件 .index / .log

2. Partiton：topics有许多分区(物理层以partition为分组)，以此来实现处理大量抽象数据的能力,每个分区对应一个文件夹，任何发布到此partition的消息都会被追加到log文件尾部所以是顺序存储(磁盘顺序写的效率比随机内存还要高)，该文件夹存储该partition的数据和索引文件
   partition名称规则-> topic名称+有序序号(0...最大序号为partition数量减1)
   2.1. 每条消息发送broker中，会根据partition规则选择被存储到哪一个partition，
   2.2. 如果能够均匀的分布在不同的partition实现了水平扩展
   2.3. 如果一个topic对应一个partition，那这个文件所在机器I/O将会成为瓶颈
   2.4. 可以指定消息的key，producer根据这个key和partition机制来判断这个消息发送到哪个partition

3. Partition offset：Each partitioned message has a unique sequence id called as offset.-唯一标识具体某一个消息
   分区理解:把数据分段划分在多个位置存放，可以是同一块磁盘也可以是不同的机器 ，一个文件夹目录？？？

4. segment：一个partition物理上由多个segment组成，消息如果在内存中放不下，就会放在文件中
   4.1 partition文件夹下，相当于一个大文件被平均分到多个相等的segment(段)，每个partition只要支持顺序读写即可，也方便已消费的信息清理，提高磁盘利用率
   4.2 seg文件由两部分组成，分别为“.index”和“.log”文件->segment索引文件和数据文件_目录中文件格式
   4.3 partition全局的第一个seg从0开始，后续每个segment文件名是一个segment文件最后一条消息的offset值，数值大小64位，20位数字字符，无数字用0补充
   参考链接： https://blog.csdn.net/lp284558195/article/details/80297208

5. Replicas of partition：Replicas are nothing but backups of a partition. Replicas are never read or write data. They are used to prevent data loss.

6. brokers ：中间件(服务器)
   Each broker may have zero or more partitions per topic.

7. Kafka Cluster：Kafka’s having more than one broker are called as Kafka cluster无需停机即可扩展Kafka群集。 这些集群用于管理消息数据的持久性和复制。

8. producers

9. consumers

10. leader： Every partition has one server acting as a leader.

11. follower

Cluster Architecture

Broker

1. Kafka brokers are stateless, so they use ZooKeeper for maintaining their cluster state
2. Kafka broker leader election can be done by ZooKeeper.
   Zookeeper
   ZooKeeper is used for managing and coordinating Kafka broker
   Producers Kafka producer doesn’t wait for acknowledgements from the broker and sends messages as fast as the broker can handle.
   Consumer group
   Consumers

WorkFlow

topic会被划分为n（n>1）个partition(水平扩展,否则存储topic的服务器可能会成为IO瓶颈)，每一条消息发送到broker时，会根据partition规则选择存储到哪一个partition
a linearly ordered sequence of messages， where each message is identified by their index (called as **offset**).

Kafka provides both pub-sub and queue based messaging system in a fast, reliable, persisted, fault-tolerance and zero downtime manner.

Pub-Sub的工作流程
1. producer发送message到topic
2. Kafka中间件储存所有的messages在n个分区上（为特定的topic划分的），保证所有的分区之间的数据是equally的，如果producer发送 two messages，这样就会有两个partitions，Kafka将会把one message存储到第一个分区，第二个消息存储到第二个分区
3. Consumer订阅一个特定的topic
4. 一旦consumer订阅了topic，Kafka将给予当前的offset of the topic 给consumer，and 储存 offset 到zookeeper ensemble中
5. consumer在常规间隔（regular interval）请求Kafka来获得new message
6. Kafka一旦从producer接收到生产消息，就会把这些消息推送到consumer
7. Consumer接收消息并处理它
8. 一旦这个消息接收并处理了，consumer会发送一个ack到Kafka broker
9. Kafka收到来自consumer的ack，它改变offset为一个新值并且在zookeeper中更新。在zookeeper中更新了offset，这样consumer能够正确读下一条消息even during server outrages.
10. 对于传统的MQ，一般会删除已经被消费的消息，kafka集群会保留，无论是否被消费，当然，因为磁盘限制，kafka提供两种策略去删除旧数据，一是基于时间，而是基于partition文件大小
11.
    
    
Queue Messaging/Consumer Group
在队列消息系统中，有一组consumer，他们有着相同的Group ID会订阅一个topic。consumers（same Group ID）订阅一个topic，被认为是同一个group，messages能在他们之间共享

kafka+zk
分布式消费的概念: 分区分段，一个区一个段只允许一个线程访问，返回横向扩展的特性
consumer group 下的consumer thread的数量等于partition数量是最优的
确认机制

offset: 记录partition要消费的消息位置，message状态

问题：

1. 消息的有序性

传统的MQ，如果多个consumers同时消费同一个服务器消费消息
服务端，按顺序发布消息，但是消息是被异步的分发到各consumer上，到达到顺序不一定，可能失去原来的顺序，为了避免这个通常使用“专用consumer”的概念，只允许一个消费者消费，失去了并发行
kafka可以在多个consumer组并发的情况下提供较好的有序性和负载均衡，将每一个分区只分发个一个consumer组
kafka只能保证一个partition-分区之内的消息的有序性

2. 为什么这么快

持久化，写入磁盘，磁盘也可以很快？

1. 页缓存技术(cache pageCache)
   页缓存就是将一个文件在内存中的所有物理页所组成的一种树形结构，我们称之为基数树，用于管理属于同一个文件在内存中的缓存内容。

2. 顺序写 由于操作系统提供了预读和写技术，磁盘的顺序写大多数情况下比随机写内存还要快，kafka避免随机的磁盘读写

3. 零拷贝技术减少拷贝次数(Zero-copy)：
   3.1、 kafka中存在大量的网络数据持久化到磁盘(producer到broker)和磁盘文件通过网络发送(broker到consumer)的过程，这一过程的性能直接影响kafka的整体吞吐量;
   3.2、 传统模式，磁盘文件->数据读入内存->socket将内存数据发送出去
   这一过程实际上发生四次数据拷贝，还伴随着四次上下文切换
   零拷贝无需cpu拷贝

4. 批量处理，合并小的请求，然后以流的方式进行交互，直至网络上线
   pull拉模式， 使用拉模式进行消息的获取消费，与消费端处理能力相符
   消费被处理状态，由consumer端维护，而不是server端
   常量时间的操作效率，读取文件而不是B树什么的
   参考链接:https://www.cnblogs.com/dw-haung/p/10393338.html

5. 数据可靠性和持久性的保证

producer向leader发送数据时，可以通过request.required.acks参数设置可靠性的级别:
1-> 保证leader已成功收到数据并得到确认后发送下一条message，如果leader宕机，则会丢失数据
0->producer无需等待broker的确认而继续发送下一批消息，这种情况下数据传输率最该，数据可靠性最低
-1->所有的follower都确认接收到数据后才算一次发送完成
at once、 at least、 exactly once

4. leader选举，leader宕机数据丢失，两个leader的情况

所有follower都成功复制才会被认为
提交，这样避免部分数据写进了leader，还没来得及被复制就宕机了

2. 如果leader不在了，新的leader必须拥有原来的leader commit的所有消息(避免followed落后太多)，但是如果一个消息被commit前等待更多的follower确认，那么他挂掉之后就有更多的follower可以成为新的leader，也会造成吞吐率的下降

5. 消息传输保障

at most once 可能会丢，不会重复
at least once 消息决不会丢，可能会重复
exactly once 每条消息肯定会被传输一次且仅传输一次
当producer向broker发送消息时，一旦这条消息被commit，由于副本(replication)的存在，它就不会丢失
但如果发送到broker就立即断网，producer就无法判断该条消息是否已被commit了，kafka此时就存在重试机制

6. 消息去重

要做到exactly once 就需要引入消息去重
GUID(globally unique identifier)得到每个消息unique id，同时可映射到broker上储存的地址
即可通过GUID便可查询提取消息内容，也便于发送方幂等性的保证
所以一般是业务方根据自身业务特点去重，比如:
分布式事务
MVCC
乐观锁

https://blog.csdn.net/jason_xiaojie_liu/article/details/81746407
https://blog.csdn.net/weixin_39861172/article/details/81906456
https://blog.csdn.net/linke1183982890/article/details/83303003