第二章：Kafka基础入门与核心概念解析

一、kafka基本概念

#1.Kafka是一款分布式消息发布和订阅系统，它的特点是高性能、高吞吐量。

#2.最早设计的目的是作为LinkedIn的活动流和运营数据的处理管道。这些数据主要是用来对用户做用户画像分析以及服务器性能数据的一些监控

#3.所以kafka一开始设计的目标就是作为一个分布式、高吞吐量的消息系统，所以适合运用在大数据传输 场景。

二、kafka特性

#1.高吞吐、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个主题 可以分多个分区, 消费组对分区进行消费操作；

#2.可扩展性：kafka集群支持热扩展；

#3.持久性、可靠性：消息可以持久化到本地磁盘，支持数据备份；

#4.容错性：允许集群中节点失败（若副本数为n，则允许n-1个节点失败）；

#5.高并发：支持数千个客户端同时读写;

三、kafka应用场景

#1.日志收集：Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如Hadoop、Hbase、Solr等；

#2.消息系统：解耦和生产者和消费者、缓存消息等；

#3.用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到Hadoop、数据仓库中做离线分析和挖掘；

#4.运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；

#5.流式处理：比如spark streaming和storm；

四、kafka技术优势

#1.可伸缩性：
Kafka 集群在运行期间可以轻松地扩展或收缩（可以添加或删除代理），而不会宕机。可以扩展一个 Kafka 主题来包含更多的分区。由于一个分区无法扩展到多个代理，所以它的容量受到代理磁盘空间的限制。能够增加分区和代理的数量意味着单个主题可以存储的数据量是没有限制的。

#2.容错性和可靠性：
Kafka 的设计方式使某个代理的故障能够被集群中的其他代理检测到。由于每个主题都可以在多个代理上复制，所以集群可以在不中断服务的情况下从此类故障中恢复并继续运行。

#3.吞吐量：
代理能够以超快的速度有效地存储和检索数据。

五、kafka基本架构

先看图，通过这张图我们来捋一捋相关的概念及之间的关系：

#1.Producer：Producer即生产者，消息的产生者，是消息的入口。

#2.kafka cluster：kafka集群结构

#3.Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，我们姑且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1等……

#4.Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。

#5.Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！

#6.Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

#7.Message：每一条发送的消息主体。

#8.Consumer：消费者，即消息的消费方，是消息的出口。

#9.Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！

#10.Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。

六、kafka工作流程分析

1.发送数据

1）写入流程

上面的架构图中，producer就是生产者，是数据的入口。Producer在写入数据的时候永远的找leader，不会直接将数据写入follower！那leader怎么找呢？写入的流程又是什么样的呢？我们看下图：

#1.先从集群获取分区的leader
#2.producter将消息发送给leader
#3.leader将消息写入本地文件，消息写入leader后，follower是主动的去leader进行同步的！
#4.followers从leader pull 消息
#5.followers将消息写入本地后向leader发送ACK

2）写入示意图

producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的！写入示意图如下：

3）分区目的

上面说到数据会写入到不同的分区，那kafka为什么要做分区呢？分区的主要目的是：
#1.方便扩展
	因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。
	
#2.提高并发
	以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。

4）分区原则

想必负载均衡这个词并不陌生，当我们向某个服务器发送请求的时候，服务端可能会对请求做一个负载，将流量分发到不同的服务器，那在kafka中，如果某个topic有多个partition，producer又怎么知道该将数据发往哪个partition呢？kafka中有几个原则：

#1.partition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。
#2.如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。
#3.如果既没指定partition，又没有设置key，则会轮询选出一个partition。

5）ACK应答机制

保证消息不丢失是一个消息队列中间件的基本保证，那producer在向kafka写入消息的时候，怎么保证消息不丢失呢？其实上面的写入流程图中有描述出来，那就是通过ACK应答机制！在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据，这个参数可设置的值为0、1、all。
#1.0
	代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。
#1.1
	代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。

#2.all
	代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。

ps:如果往不存在的topic写数据，能不能写入成功呢？kafka会自动创建topic，分区和副本的数量根据默认配置都是1。

2.保存数据

Producer将数据写入kafka后，集群就需要对数据进行保存了！kafka将数据保存在磁盘，可能在我们的一般的认知里，写入磁盘是比较耗时的操作，不适合这种高并发的组件。Kafka初始会单独开辟一块磁盘空间，顺序写入数据（效率比随机写入高）。

1）Partition 结构

前面说过了每个topic都可以分为一个或多个partition，如果你觉得topic比较抽象，那partition就是比较具体的东西了！Partition在服务器上的表现形式就是一个一个的文件夹，每个partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、.log文件、.timeindex文件（早期版本中没有）三个文件， log文件就实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。

如上图，这个partition有三组segment文件，每个log文件的大小是一样的，但是存储的message数量是不一定相等的（每条的message大小不一致）。文件的命名是以该segment最小offset来命名的，如000.index存储offset为0~368795的消息，kafka就是利用分段+索引的方式来解决查找效率的问题。

2）Message结构

上面说到log文件就实际是存储message的地方，我们在producer往kafka写入的也是一条一条的message，那存储在log中的message是什么样子的呢？消息主要包含消息体、消息大小、offset、压缩类型……等等！我们重点需要知道的是下面三个：

1.offset：offset是一个占8byte的有序id号，它可以唯一确定每条消息在parition内的位置！
2.消息大小：消息大小占用4byte，用于描述消息的大小。
3.消息体：消息体存放的是实际的消息数据（被压缩过），占用的空间根据具体的消息而不一样。

3）存储策略

无论消息是否被消费，kafka都会保存所有的消息。那对于旧数据有什么删除策略呢？
1.基于时间，默认配置是168小时（7天）。
2.基于大小，默认配置是1073741824。

启用删除策略
log.cleanup.policy=delete
#专门的日志删除任务来周期性检测和删除不符合保留条件的日志分段文件，默认300000ms，5分钟
log.retention.check.interval.ms=300000
#可配置以下两个策略：
#1、清理超过指定时间清理： 
log.retention.hours=16
#2、超过指定大小后，删除旧的消息;默认值为-1，表示无穷大；
log.retention.bytes=1073741824
 
#注意：
#log.retention.bytes和log.retention.hours任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖
kafka读取特定消息的时间复杂度是O(1)，所以这里删除过期的文件并不会提高kafka的性能！

3.消费数据

消息存储在log文件后，消费者就可以进行消费了。在讲消息队列通信的两种模式的时候讲到过点对点模式和发布订阅模式。Kafka采用的是点对点的模式，消费者主动的去kafka集群拉取消息，与producer相同的是，消费者在拉取消息的时候也是找leader去拉取。

多个消费者可以组成一个消费者组（consumer group），每个消费者组都有一个组id！同一个消费组者的消费者可以消费同一topic下不同分区的数据，但是不会组内多个消费者消费同一分区的数据！！！是不是有点绕。我们看下图：

图示是消费者组内的消费者小于partition数量的情况，所以会出现某个消费者消费多个partition数据的情况，消费的速度也就不及只处理一个partition的消费者的处理速度！如果是消费者组的消费者多于partition的数量，那会不会出现多个消费者消费同一个partition的数据呢？上面已经提到过不会出现这种情况！多出来的消费者不消费任何partition的数据。所以在实际的应用中，建议消费者组的consumer的数量与partition的数量一致！

在保存数据的小节里面，我们聊到了partition划分为多组segment，每个segment又包含.log、.index、.timeindex文件，存放的每条message包含offset、消息大小、消息体……我们多次提到segment和offset，查找消息的时候是怎么利用segment+offset配合查找的呢？假如现在需要查找一个offset为368801的message是什么样的过程呢？我们先看看下面的图：

1.先找到offset的368801message所在的segment文件（利用二分法查找），这里找到的就是在第二个segment文件。

2.打开找到的segment中的.index文件（也就是368796.index文件，该文件起始偏移量为368796+1，我们要查找的offset为368801的message在该index内的偏移量为368796+5=368801，所以这里要查找的**相对offset**为5）。由于该文件采用的是稀疏索引的方式存储着相对offset及对应message物理偏移量的关系，所以直接找相对offset为5的索引找不到，这里同样利用二分法查找相对offset小于或者等于指定的相对offset的索引条目中最大的那个相对offset，所以找到的是相对offset为4的这个索引。

3.根据找到的相对offset为4的索引确定message存储的物理偏移位置为256。打开数据文件，从位置为256的那个地方开始顺序扫描直到找到offset为368801的那条Message。

这套机制是建立在offset为有序的基础上，利用segment+有序offset+稀疏索引+二分查找+顺序查找等多种手段来高效的查找数据！至此，消费者就能拿到需要处理的数据进行处理了。那每个消费者又是怎么记录自己消费的位置呢？在早期的版本中，消费者将消费到的offset维护zookeeper中，consumer每间隔一段时间上报一次，这里容易导致重复消费，且性能不好！在新的版本中消费者消费到的offset已经直接维护在kafk集群的__consumer_offsets这个topic中！