2018年第52周-Kafka知识点

预祝2019年元旦节快乐！
2018年最后一周，分享些Kafka的知识点。

Kafka数据分区和消费者的关系

分区（partition）

topic是逻辑概念，分区（partition）是物理概念，对于用户来讲是透明的。producer只须要关心消息网哪一个topic发送，而consumer之关系本身订阅哪一个topic，不须要关心每条消息存于整个集群的哪一个Broker。

为了性能考虑，若是topic的消息都放在一个Broker，这个Broker必然称为瓶颈，并且没法作到水平扩展。因此topic内的数据分布到整个集群就是个天然而然的设计了。分区的引入就是解决水平扩展问题的一个解决方案。

Kafka尽可能将全部的分区均匀分配到整个集群上。基本算法以下：

1. 将全部存货的N个Broker和待分配的分区排序。
2. 将第i个分区分配到第（i mon n）个Broker上，这个分区的第一个副本（Replica）存在于这个分配的Broker上，而且会做为分区的优先副本。
3. 将第i个分区的第j个副本分配到第（i+j）mod n 个Broker上。

但实际状况Kafka的算法是上述基础上再加些，看Kafka的函数assignReplicasToBrokers。变了两点：

1. 限制了副本因子（replication factor）不得大于Broker的个数。由于当j大于n时，就会存在一个Broker有两个副本，这没意义且浪费。
2. 起始位置不是第0个Broker，是第rand.nextInt(Brokers.size)个。那是由于（i mon n）形成的问题就是，永远都会有0，未必有n。因此必须加上随机数，也就是第i个分区分配到第（i mon n）+rand.nextInt(Brokers.size)个Broker上。
3. 若是考虑多机架问题，那么Broker顺序就未必是0,1,2,3,4,5。而是若是0，1,2是机架A，3,4,5是机架B。则Broker的顺序为0，3，1,4,2,5。错开，每一个机架依次选一次。因此当副本因子为3时，保证每个分区在各个机架都至少有一个副本。

分区中副本通常都是 Leader，其他的都是 Follow 副本。生产者消费者都固定在 Leader进行生产和消费。

分区与生产者

　负载均衡（Load balancing）　　

生产者直接发送数据到Broker，不须要任何的中间路由层，而接受的Broker是该分区的Leader。
为了帮助生产者实现这一点，全部Broker均可以回答关于哪些是可用服务器的元数据的请求，以及在任何给定的时间内，某个主题的分区的Leader是否容许生产者适当地发送它的请求。
客户端能够控制往哪一个分区生产消息。这能够随机地进行，实现一种随机的负载平衡，或者能够经过一些语义分区函数来实现负载平衡。
Kafka提供了语义分区的接口，容许用户指定一个分区的key，并使用这个key来作hash到一个分区(若是须要的话，也是能够复写这分区功能的)。例如，咱们选择user的id做为可用，则因此该用户的信息都会发送到一样的分区。

异步发送（Asynchronus send）

批处理是效率的主要驱动因素之一，为了可以批处理，Kafka的生产者会尝试在内存中积累数据，而后在一块儿在一个请求中以大批量的形式发送出去。批处理这个能够设置按固定的消息数量或按特定的延迟（64k或10ms）。这容许累积更多字节的发送出去，这样只是在服务器上作少许的大IO操做。这种缓冲是可配置的，这样提供了一种机制来以额外的延迟来提升吞吐量。具体的配置)和生产者的api能够在这文档中找到。

分区与消费者

消费者的工做方式是，向分区的Leader发送“fetch”请求。在每一个请求中消费者指定日志的偏移量（position），而后接受回一大块从偏移量开始的日志。所以，消费者对偏移量有重要的控制权，若是须要，能够重置偏移量来从新消费数据。

Push和pull

咱们首先考虑的一个问题是，消费者应该是从Broker拉取消息，仍是应该是Broker把消息推送给消费者。在这方面，Kafka遵循了一种更传统的设计，大多数消息队列系统也会用的，那就是数据是从生产者push到Broker，消费者是从Broker拉取数据。一些日志集中系统，如Scribe和Apache Flume，遵循一个很是不一样的，基于推送的路径，将数据被推到下游。这两种方法都由利弊，在基于推送的系统，因为是Broker得控制数据传输的速率，不一样消费者可能要不一样的速率。然而消费者通常的目的都是让消费者本身可以以最大的速度进行消费，但在基于push的系统，当消费速率低于生产效率时，消费者就不知道该怎么办好了（本质上就是一种拒绝服务攻击（DOS））。一个基于pull的系统就拥有很好的熟悉，消费者能够简单的调控速率。

基于pull的系统的另外一个优势是，它能够对发送给消费者的数据进行聚合的批处理。基于推送的系统必须选择当即发送请求或积累更多数据，而后在不知道下游用户是否可以当即处理它的状况下发送它。

基于pull的系统的缺点是，若是Broker没数据，则消费者可能会不停的轮训。为了不这一点，咱们在pull请求上提供了参数，容许消费者在“长轮训”中阻塞，直到数据达到（而且能够选择等待，直到必定数量的本身能够，确保传输的大小）。

消费者的Position（Consumer Position）

使人惊讶的是，跟踪消息是否使用了，是消息队列系统的关键性能点之一。
不少消息队列系统在Broker中保存了关于什么消息是被消费了的元数据。也就是说，当消息队列给消费者时，Broker要么当即记录信息到本地，要么就是等待消费者的确认。这是一个至关直观的选择，并且对于一台机器服务器来讲，很清楚地知道这些消息的状态。因为许多消息队列系统中用于存储的数据结构都很糟糕，所以这（记录消息状态）也是一个实用的选择——由于Broker知道什么是已经被消费的，因此能够当即删除它，保持数据的大小。

让Broker和消费者就已经消费的东西达成一致，这可不是小问题。若是一条消息发送到网络上，Broker就把它置为已消费，但消费者可能处理这条消息失败了（或许是消费者挂了，也或许是请求超时等），这条消息就会丢失了。为了解决这个问题，不少消息队列系统增长了确认机制。当消息被发送时，是被标志为已发送，而不是已消费；这是Broker等待消费者发来特定的确认信息，则将消息置为已消费。这个策略虽然解决了消息丢失的问题，但却带来了新的问题。第一，若是消费者在发送确认信息以前，在处理完消息以后，消费者挂了，则会致使此消息会被处理两次。第二个问题是关于性能，Broker必须保存每一个消息的不一样状态（首先先锁住消息以至于不会让它发送第二次，其次标志位已消费从而能够删除它）。还有些棘手的问题要处理。如消息被发送出去，但其确认信息一直没返回。

Kafka处理则不同。咱们的主题被分为一个有序分区的集合，且每一个分区在任何给定的时间内只会被订阅它的消费者组中的一个消费者给使用。这意味着每一个分区中的消费者的position仅仅是一个整数，这是下一次消费时，消息的偏移量。这使状态（记录是否被消费）很是小，每一个分区只有一个数字。这个状态能够被按期检查。这样确认一条消息是否被消费的成本就很低。

这样还附加了一个好处。消费者能够重置其最早的position从而从新消费数据。这虽然违反了队列的公共契约，但它却变成关键功能给许多消费者。例如，若是消费者代码有一个bug，而且在一些消息被消费后才被发现，那么当bug被修复后，消费者就能够从新使用这些消息。

消费组

每群消费者都会被标志有消费组名。有消费组这个概念，Kafka就能够实现相似与工做队列（Worke Queues）模式和发布/订阅（Publish/Subscribe）。
若是消费者都在同一个消费组，则消息则会负载均衡的分配每一个消费者，一条消息不会分配个两个及以上的消费者。
若是消费者不在同一个组，则消息会被广播到每个消费组中。

Kafka的数据offset读取流程

每一个消息在分区中都是被分配一个有序的ID数字，而这数字，咱们称之为偏移量（offset）。在一个分区上，offset惟一标识一个消息。
由每一个消费者维护offset。

在Kafka文件存储中，同一个topic下有多个不一样分区，每一个分区为一个目录，分区命名规则为topic名称+有序序号，第一个分区序号从0开始，序号最大值为分区数量减1。

partition物理上由多个大小相等的segment组成。segment由2大部分组成，分别为index file和data file，此2个文件一一对应，成对出现，后缀".index"和“.log”分别表示为segment索引文件、数据文件.
segment文件命名规则：partion全局的第一个segment从0开始，后续每一个segment文件名为上一个segment文件最后一条消息的offset值。数值最大为64位long大小，19位数字字符长度，没有数字用0填充。

00000000000000000.index
00000000000000000.log
00000000000368769.index
00000000000368769.log
00000000000737337.index
00000000000737337.log
00000000001105814.index
00000000001105814.log

index file的结构：

1,0
3,497
6,1407
8,1686
....
N,position

index file结构是两个数字两个数字一组，N,position。N用于查找相对于当前文件名的offset值的N个消息。如00000000000368769.index的3,497，则为368769+3=第368772个消息。而position 497是指data file的偏移量497。

data file由许多message组成，message物理结构以下：

8 byte offset
4 byte message size
4 byte CRC32
1 byte "magic"
1 byte "attributes"
4 byte key length
K byte key
4 byte payload length
value bytes payload

这样的结构，配合index file，很快就能够知道某条消息的大小。

　在partition中如何经过offset查找message

例如读取offset=368776的message，须要经过下面2个步骤查找。

第一步查找segment file

上述为例，其中00000000000000000000.index表示最开始的文件，起始偏移量(offset)为0.第二个文件00000000000000368769.index的消息量起始偏移量为368770 = 368769 + 1.一样，第三个文件00000000000000737337.index的起始偏移量为737338=737337 + 1，其余后续文件依次类推，以起始偏移量命名并排序这些文件，只要根据offset 二分查找文件列表，就能够快速定位到具体文件。
当offset=368776时定位到00000000000000368769.index|log

第二步经过segment file查找message

经过第一步定位到segment file，当offset=368776时，依次定位到00000000000000368769.index的元数据物理位置和00000000000000368769.log的物理偏移地址，而后再经过00000000000000368769.log顺序查找直到offset=368776为止。

从上述图3可知这样作的优势，segment index file采起稀疏索引存储方式，它减小索引文件大小，经过mmap能够直接内存操做，稀疏索引为数据文件的每一个对应message设置一个元数据指针,它比稠密索引节省了更多的存储空间，但查找起来须要消耗更多的时间。

Kafka高效文件存储设计特色

• Kafka把topic中一个parition大文件分红多个小文件段，经过多个小文件段，就容易按期清除或删除已经消费完文件，减小磁盘占用。
• 经过索引信息能够快速定位message和肯定response的最大大小。
• 经过index元数据所有映射到memory，能够避免segment file的IO磁盘操做。
• 经过索引文件稀疏存储，能够大幅下降index文件元数据占用空间大小。

注： 稀疏索引相似于带一级索引的跳表，可是一级索引是数组可使用二分法查找。

注：mmap()函数是Linux的文件空间映射函数，用来将文件或设备空间映射到内存中，能够经过映射后的内存空间存取来得到与存取文件一致的控制方式，没必要再使用read(),write()函数。
mmap和常规文件操做的区别
回顾一下常规文件系统操做（调用read/fread等类函数）中，函数的调用过程：

1. 进程发起读文件请求。
2. 内核经过查找进程文件符表，定位到内核已打开文件集上的文件信息，从而找到此文件的inode。
3. inode在address_space上查找要请求的文件页是否已经缓存在页缓存中。若是存在，则直接返回这片文件页的内容。
4. 若是不存在，则经过inode定位到文件磁盘地址，将数据从磁盘复制到页缓存。以后再次发起读页面过程，进而将页缓存中的数据发给用户进程。

总结来讲，常规文件操做为了提升读写效率和保护磁盘，使用了页缓存机制。这样形成读文件时须要先将文件页从磁盘拷贝到页缓存中，因为页缓存处在内核空间，不能被用户进程直接寻址，因此还须要将页缓存中数据页再次拷贝到内存对应的用户空间中。这样，经过了两次数据拷贝过程，才能完成进程对文件内容的获取任务。写操做也是同样，待写入的buffer在内核空间不能直接访问，必需要先拷贝至内核空间对应的主存，再写回磁盘中（延迟写回），也是须要两次数据拷贝。
而使用mmap操做文件中，建立新的虚拟内存区域和创建文件磁盘地址和虚拟内存区域映射这两步，没有任何文件拷贝操做。而以后访问数据时发现内存中并没有数据而发起的缺页异常过程，能够经过已经创建好的映射关系，只使用一次数据拷贝，就从磁盘中将数据传入内存的用户空间中，供进程使用。
总而言之，常规文件操做须要从磁盘到页缓存再到用户主存的两次数据拷贝。而mmap操控文件，只须要从磁盘到用户主存的一次数据拷贝过程。说白了，mmap的关键点是实现了用户空间和内核空间的数据直接交互而省去了空间不一样数据不通的繁琐过程。所以mmap效率更高。

函数原型

void *mmap(void *start, size_t length, int prot, int flags, int fd, off_t offset);

也就是能够将大数据的文件，局部映射到内存中，在内存中进行此部分文件的操做。对此内存操做，都不涉及到内核空间到用户空间之间交互。直接操做内存，内存直接写入（读取）文件。就只有一次IO。 若是是普通文件操做，则须要文件复制到内核，再由内核复制到用户空间，用户空间才能操做。从而达到零拷贝。
换句话说，但凡是须要用磁盘空间代替内存的时候，mmap均可以发挥其功效。

Kafka内部如何保证顺序，结合外部组建如何保证消费者的顺序

Kafka中每一个分区都是有序，因为Kafka的消息是不可变的，因此都是追加的形式有序的往上加消息。这个结构体叫 结构化提交日志（a structured commit log）。

首先就要考虑是否真的须要全部消息在队列都得有序。通常状况，不止通常，而是很大一部分，是能够无序的。就跟分布式同样。有不少业务，看起来是同步的，静下来慢慢思考，就会发现不少东西是能够异步执行的。
若是实在有这样保证顺序的须要，保证生产者需将有序地提交给一个分区，首先是生产者不能提交错顺序。其次，消费者组就不能拥有两个或以上消费者实例了。连两个或以上的消费者组也不能有。

持久化

Kafka会根据保留时间这参数，持久化全部已经收到的消息。虽然能够设置保留时间这参数，可是Kafka优秀的性能，添加删除都是常量级的性能，因此理论上，数据保存很长时间也不成问题。

参考:
http://kafka.apache.org/
https://www.zhihu.com/questio...
https://blog.csdn.net/yangyut...
http://www.cnblogs.com/huxiao...
https://www.cnblogs.com/ITtan...

稀疏索引：https://blog.csdn.net/qq_2223...
跳表：https://www.jianshu.com/p/dc2...