大数据之Kafka框架

朱雀 2022-12-26 10:28 179阅读 0赞

# 大数据之Kafka框架 #

### 目录 ###

*  大数据之Kafka框架
 *   *  第1章 Kafka概述
     *   *  1.1 消息队列（Message Queue）
         *  1.2 定义
         *  1.3 Kafka基础架构
     *  第2章 Kafka快速入门
     *   *  2.1 Kafka命令行操作
     *  第3章 Kafka架构深入
     *   *  3.1 Kafka工作流程及文件存储机制
         *  3.2 Kafka生产者
         *   *  3.2.1 分区策略
             *  3.2.2 数据可靠性保证
             *  3.2.3 follower与leader故障之后的解决方案
             *  3.2.4 Exactly Once语义
         *  3.3 Kafka消费者
         *   *  3.3.1 消费方式
             *  3.3.2 分区分配策略
             *  3.3.3 offset的维护
         *  3.4 Kafka高效读写数据
         *  3.5 Zookeeper在Kafka中的作用
         *  3.6 Kafka事务
         *   *  3.6.1 Producer事务
             *  3.6.2 Consumer事务（精准一次性消费）
         *  3.7 Producer 消息发送流程

## 第1章 Kafka概述 ##

### 1.1 消息队列（Message Queue） ###

**1.1.1 传统消息队列的应用场景**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

**1.1.2 消息队列的两种模式**

> **1）点对点模式**（**一对一**，消费者主动拉取数据，消息收到后消息清除）  
> 消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。  
> 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。  
> **2）发布/订阅模式**（**一对多**，消费者消费数据之后不会清除消息）  
> 消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。

### 1.2 定义 ###

Kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理领域。

### 1.3 Kafka基础架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

> 1.  producer:生产者,向kafka中写入消息
> 2.  consumer:消费者,用于向kafka中读取消息
> 3.  Topic: 主题,一般工作中是一个业务一个主题
> 4.  partition: Topic的分区,为了实现分布式存储,提高并行度,加快读取写入的速度
> 5.  broker: kafka的一个节点，用于存放分区数据
> 6.  副本: kafka为了提高partition数据的安全性,提供了副本机制
> 7.  leader: 副本的主节点,producer发送消息以及consumer消息消息都是找leader
> 8.  follower: 副本的从节点,follower主要用于从leader同步消息,但是如果leader挂了,会从follower中选举出新leader
> 9.  consumer group: 消费者组,组中有多个消费者,一般一个消费者组用于消费一个Topic,所以消费者组中的消费者的个数最好=topic的分区数，消费者组消费一个topic的时候,topic的一个分区的数据只能被组中一个消费者所消费
> 10. zookeepr: 主要监控broker的状态,leader选举也依赖zookeeper
> 11. offset: producer写到分区的每条数据都有一个唯一标识,这个唯一标识就是offset。
> 12. 消费者消费分区的数据的时候,如果分区的数据特别多,消费者不能一次消费完分区数据,所以需要分批次进行消费,所以在开始下一个批次开始消费的时候必须知道本次应该从哪个offset开始消费。消费者每次消费完数据之后都会记录\[记录在\_\_offset\_consumer这个内部topic中\]下一次应该从哪里开始消费

## 第2章 Kafka快速入门 ##

### 2.1 Kafka命令行操作 ###

1）查看当前服务器中的所有topic

[hadoop@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka --list

2）创建topic

[hadoop@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka \
    --create --replication-factor 3 --partitions 1 --topic first
    选项说明：
    --topic 定义topic名
    --replication-factor  定义副本数
    --partitions  定义分区数

3）删除topic

[hadoop@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka \
    --delete --topic first
    需要server.properties中设置delete.topic.enable=true否则只是标记删除。

4）发送消息

[hadoop@hadoop102 kafka]$ bin/kafka-console-producer.sh \
    --broker-list hadoop102:9092 --topic first
    >hello world
    >atguigu  atguigu

5）消费消息

[hadoop@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
    --bootstrap-server hadoop102:9092 --from-beginning --topic first
    
    [hadoop@hadoop103 kafka]$ bin/kafka-console-consumer.sh \
    --bootstrap-server hadoop102:9092 --from-beginning --topic first
    --from-beginning：会把主题中以往所有的数据都读取出来。

6）查看某个Topic的详情

[hadoop@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka \
    --describe --topic first

7）修改分区数

[hadoop@hadoop102 kafka]$bin/kafka-topics.sh --zookeeper hadoop102:2181/kafka --alter --topic first --partitions 6

## 第3章 Kafka架构深入 ##

### 3.1 Kafka工作流程及文件存储机制 ###

> Kafka中消息是以topic进行分类的，生产者生产消息，消费者消费消息，都是面向topic的。  
> topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，且每条数据都有自己的offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。

文件存储机制
    		Topic:[逻辑上的概念]
    			partition: [物理上真实存在,位置在log.dir参数配置的地方,名称=topic名称-分区号]
    				segment:
    					index文件： log文件数据的索引
    					log文件： 存储的数据
    					timestampindex文件: log文件数据的时间索引
    		partition下segment的文件名命名规则:
    			第一个segment的文件名为00000000000000000000
    			第N个segment的文件名=第N-1个segment文件最后一个offset+1

> 如何根据offset找到对应的数据?  
> 1、利用二分查找法,根据segment的文件名确定offset在哪个segment  
> 2、根据index文件确定offset在log文件哪个区间  
> 3、扫描log文件的区间获取数据

### 3.2 Kafka生产者 ###

#### 3.2.1 分区策略 ####

**1）分区的原因**

> （1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；  
> （2）可以提高并发，因为可以以Partition为单位读写了。  
> **2）分区的原则**  
> 我们需要将producer发送的数据封装成一个ProducerRecord对象。
> 
> > 1、直接指定分区号: 直接将消费发送到对应的分区  
> > 2、没有指定分区号,但是有指定key： 将消息发到对应的分区\[key.hashCode % 分区数\]  
> > 3、没有指定分区号,也没有key  
> > 新版本:  
> > 发送第一个批次的时候会生成一个随机数,分区号=随机数%分区数,将数据发到对应的分区  
> > 发送第N个批次的时候,从所有的分区中排除掉上一次发送的分区，从剩下的分区中随机选择一个发送  
> > 旧版本:  
> > 发送第一个批次的时候会生成一个随机数,分区号=随机数%分区数,将数据发到对应的分区  
> > 发送第N个批次的时候，会将数据发送到 (第一个批次生成的随机数+(N-1)) % 分区数 所在的分区

#### 3.2.2 数据可靠性保证 ####

> 为保证producer发送的数据，能可靠的发送到指定的topic，topic的每个partition收到producer发送的数据后，都需要向producer发送ack（acknowledgement确认收到），如果producer收到ack，就会进行下一轮的发送，否则重新发送数据。

> kafka通过ack机制来保证数据的可靠性  
> ack=0: leader接受到消息之后立即返回确认消息给producer  
> 此时可能出现数据丢失的问题  
> 出现的原因: leader返回确认消息之后，在数据落盘之前宕机,数据就丢失了

> ack=1: leader接受到消息并且落盘之后才会返回确认消息给producer  
> 此时也可能出现数据丢失的问题  
> 出现原因: leader落盘数据之后返回确认消息给producer，此时leader宕机，从follower中选举出新leader,此时新leader中没有该数据

> ack=-1: leader接受到消息并且落盘，并且所有的follower都同步了消息之后才会返回确认消息给producer  
> 此时可能会出现数据重复的问题  
> 出现的原因: follower已经从leader同步了数据，但是在leader返回确认之前宕机，从follower中选举出新leader，因为producer没有收到确认消息，所以会将数据重新发送一次,此时心leader中有两条重复的数据

> 因为kafka ack=-1 副本同步数据的时候采用所有副本都同步成功才会返回确认消息，所以此时如果其中一个follower因为网络之类的原因就可能导致同步迟迟完成不了，此时kafka会将该follower踢出ISR列表。  
> ISR: 与leader同步到了一定程度\[LEO>=HW\]的follower  
> LEO: 每个副本的最后一个offset  
> HW：所有副本中最小的LEO

#### 3.2.3 follower与leader故障之后的解决方案 ####

> 1、follower故障: follower正常之后，会清除宕机之前的HW之后所有数据，然后重新从leader进行同步，一直到LEO>=ISR的HW才会重新加入ISR列表  
> 2、leader故障: 从follower中重新选举出新leader，其余的所有的follower都需要清除掉HW之后的所有数据重新从新leader同步数据

#### 3.2.4 Exactly Once语义 ####

> 三种容错语义：  
> 1、at-lest-once: 数据最少一条\[数据可能重复\]  
> 2、at-most-once: 数据最多一条\[可能有数据丢失\]  
> 3、exactly-once: 数据有且仅有一套

> kafka如何保证数据有且仅有一条？(kafka是借鉴mysql的主键思想。)  
> producer在每次发送数据的时候都会带上数据的唯一主键\[producerid+partition+sequenceNumber\],主键会缓存到broker,每次producer数据发送过来的时候都会与缓存的主键进行对比，如果缓存中主键已经存在代表数据已经发送过，那么本次数据就标记为无效
> 
> > producerid: 生产者的唯一标识,在producer端生成  
> > partition: 分区号  
> > sequenceNumber: 发到分区的第几条数据

> kafka要想实现exactly once有前提:  
> 1、ack=-1  
> 2、必须要求producer不挂，因为producer宕机之后，重启会重新生成producerid,此时再发送数据的时候主键就与宕机之前不一样，就不能确保主键唯一导致数据重复

### 3.3 Kafka消费者 ###

#### 3.3.1 消费方式 ####

**consumer采用pull（拉）模式从broker中读取数据。**

> **push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的**。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。  
> **pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据**。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

#### 3.3.2 分区分配策略 ####

> 一个consumer group中有多个consumer，一个 topic有多个partition，所以必然会涉及到partition的分配问题，即确定那个partition由哪个consumer来消费。  
> Kafka有两种分配策略，一是roundrobin，一是range。

**1）roundrobin**

比如: 
    	Topic[partition0、partition1、partition2、partition3、partition4]
    	consumer group[consumer1、consumer2]
    		分配结果:
    			consumer1消费partition0、partition2、partition4
    			consumer2消费partition1、partition3

**2）range**

计算步骤:
    	1、评估每个消费者大概消费几个分区[分区数/消费者个数]
    	2、指定前几个消费者多消费1个分区[分区数%消费者个数]
    		比如: 
    			Topic[partition0、partition1、partition2、partition3、partition4]
    			consumer group[consumer1、consumer2]
    				1、评估每个消费者大概消费几个分区[5/2=2]
    				2、指定前几个消费者多消费1个分区[5%2=1]
    		分配结果:
    			consumer1消费partition0、partition1、partition2
    			consumer2消费partition3、partition4

#### 3.3.3 offset的维护 ####

> 由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。  
> Kafka 0.9版本之前，consumer默认将offset保存在Zookeeper中，从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为\_\_consumer\_offsets。

### 3.4 Kafka高效读写数据 ###

**1）分区:分区能够提高读写的并发，从而提高速度  
2）顺序写磁盘**

> Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到到600M/s，而随机写只有100k/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

**3）应用Pagecache**

1.  减少与磁盘的交互的次数\[有了pagecache之后，可以在pagecache中存储多个批次的数据之后然后再写入磁盘\]
2.  如果网络够好\[消费者的速度=生产者的速度\]此时数据刚刚写入pagecache中,消费者就可以直接从pagecache中将数据消费走,不用经过磁盘
3.  pagecache不在JVM中，不能增加GC的负担
4.  减少磁头寻址的时间\[因为pagecache是broker对应的内存区域,生产者向broker写数据的时候首先写入pagecache中,如果pagecache中有多个topic多个分区的数据,此时会在pagecache中对数据存储的物理地址进行排序，顺序写入\]

**4）零拷贝**

内存分为两块: 内核区、用户区
    从本地磁盘读取文件写入网络,正常流程:
    	1、通过IO流将数据读取到内核区的pagecache中
    	2、将数据从内核区的pagecache中读取到用户区
    	3、Application在用户区的内存中对数据进行处理[过滤、切割..]
    	4、将处理之后的数据写入内核区的socket缓冲区
    	5、将数据从socket缓冲区写入网卡
    零拷贝流程:
    	1、通过IO流将数据读取到内核区的pagecache中
    	2、将数据从内核的pagecache中写入网卡

### 3.5 Zookeeper在Kafka中的作用 ###

> kafka监控broker节点是否正常、leader选举等都依赖zookeeper

**leader选举流程:**

1.  在启动broker的过程中,会从broker中选举一个作为controller
2.  broker在启动的过程中，会到zookeeper\[/brokers/ids\]中进行注册,controller会监控/brokers/ids从而达到监控broker的目的
3.  broker中保存的分区在broker启动之后也会到zookeeper中进行注册\[/brokers/topics/topic名称/partitions/分区号/state\],controller也会监控该节点
4.  如果broker宕机,/brokers/ids下属于该broker的节点消失,controller因为一直监控该目录,所以能够立即知道broker宕机
5.  从follower中选举出新leader，controller会更新/brokers/topics/topic名称/partitions/分区号/state

### 3.6 Kafka事务 ###

> Kafka从0.11版本开始引入了事务支持。事务可以保证Kafka在Exactly Once语义的基础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败。

#### 3.6.1 Producer事务 ####

> 为了实现跨分区跨会话的事务，需要引入一个全局唯一的Transaction ID，并将Producer获得的PID和Transaction ID绑定。这样当Producer重启后就可以通过正在进行的Transaction ID获得原来的PID。

> 为了管理Transaction，Kafka引入了一个新的组件Transaction Coordinator。Producer就是通过和Transaction Coordinator交互获得Transaction ID对应的任务状态。Transaction Coordinator还负责将事务所有写入Kafka的一个内部Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。

#### 3.6.2 Consumer事务（精准一次性消费） ####

> 上述事务机制主要是从Producer方面考虑，对于Consumer而言，事务的保证就会相对较弱，尤其时无法保证Commit的信息被精确消费。这是由于Consumer可以通过offset访问任意信息，而且不同的Segment File生命周期不同，同一事务的消息可能会出现重启后被删除的情况。

> 如果想完成Consumer端的精准一次性消费，那么需要kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将kafka的offset保存到支持事务的自定义介质中（比如mysql）。

### 3.7 Producer 消息发送流程 ###

1.  在main方法中会创建producer客户端
2.  会将发送的消息进行封装,封装成PrducerRecord对象,通过send方法将数据进行发送
3.  通过拦截器对数据进行处理
4.  通过序列化器对key、value进行序列化
5.  通过分区器对数据进行分区，决定数据应该发到哪个分区
6.  将数据写到Accumulator共享变量中，等到该topic该分区满一个批次之后才会通过sender线程将这一个批次的数据发到broker  
    整个流程中需要两个线程: main线程、sender线程

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221120/3762fd10978545428adeec7f5b3adaca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221120/3d9b42d8ec2c4f889386b206496d00b0.png