[大数据]3.Kafka

1.定义

1.1传统定义

1. Kafka是一个分布式的基于发布订阅模式的消息队列，主要应用于大数据的实时处理领域
2. 发布订阅：消息发布者不会直接将消息发送给订阅者，而是将发布的消息分为不同的类别，订阅者只接收感兴趣的消息

1.2最新定义

1. Kafka是一个开源的分布式流平台，被数千家公司用于高性能数据管道、流分析、数据集成、和关键任务应用

2.消息队列

2.1常见消息队列

1. Kafka
2. ActiveMQ
3. RabbitMQ
4. RocketMQ

2.2应用场景

1. Kafka更多应用于大数据场景
2. ActiveMQ、RabbitMQ、RocketMQ多应用于web服务器开发场景

2.3作用

1. 缓存
2. 流量削峰
3. 解耦
   
4. 异步通信

3.Kafka模式

3.1模式图鉴

1. 点对点模式

   • 消费者主动拉取数据，数据收到后清除数据

   

2. 发布/订阅模式

   • 可以有多个Topic主题
   • 消费者消费数据后，不删除数据
   • 每个消费者相互独立、都可以消费到数据
     

3.2推拉模式各自优缺点

3.2.1推模式

1. 优点
   • 1.消息实时性高， Broker 接受完消息之后可以立马推送给 Consumer。
   • 2.对于消费者使用来说更简单，只需要等着，反正有消息来了就会推过来。
2. 缺点
   • 1.推送速率难以适应消费速率，推模式的目标就是以最快的速度推送消息，
     当生产者往 Broker 发送消息的速率大于消费者消费消息的速率时，
     随着时间的增长消费者那边可能就“爆仓”了，因为根本消费不过来啊。
     当推送速率过快就像 DDos 攻击一样消费者就傻了。
   • 2.并且不同的消费者的消费速率还不一样，身为 Broker 很难平衡每个消费者
     的推送速率，如果要实现自适应的推送速率那就需要在推送的时候消费者告诉
     Broker ，我不行了你推慢点吧，然后 Broker 需要维护每个消费者的状态进行
     推送速率的变更。这其实就增加了 Broker 自身的复杂度。
3. 适用场景
   • 适用于消息量不大、消费能力强要求实时性高的情况下

3.2.2拉模式

1. 优点

   • 1.消费者可以根据自身的情况来发起拉取消息的请求。假设当前消费者觉得自己消费不过来了，它可以根据一定的策略停止拉取，或者间隔拉取都行。
   • 2.拉模式下 Broker 就相对轻松了，它只管存生产者发来的消息，至于消费的时候自然由消费者主动发起，来一个请求就给它消息呗，从哪开始拿消息，拿多少消费者都告诉它，它就是一个没有感情的工具人，消费者要是没来取也不关它的事。

2. 缺点

   • 1.消息延迟，毕竟是消费者去拉取消息，但是消费者怎么知道消息到了呢？所以它只能不断地拉取，但是又不能很频繁地请求，太频繁了就变成消费者在攻击 Broker 了。因此需要降低请求的频率，比如隔个 2 秒请求一次，你看着消息就很有可能延迟 2 秒了。
   • 2.消息忙请求，忙请求就是比如消息隔了几个小时才有，那么在几个小时之内消费者的请求都是无效的，在做无用功。
   • 3.服务端压力变大，长时间消费速度跟不上生产速度，就会使得Broker存储的消息越来越多，极端情况下可能导致服务器内存爆仓

3. 适用场景

   • 拉模式可以更合适的进行消息的批量发送，基于推模式可以来一个消息就推送，也可以缓存一些消息之后再推送，但是推送的时候其实不知道消费者到底能不能一次性处理这么多消息。而拉模式就更加合理，它可以参考消费者请求的信息来决定缓存多少消息之后批量发送。

4.架构

4.1基础架构

4.2完整架构

1. 为方便扩展，提高吞吐量，一个topic分为多个partition
2. 配合分区设计，提出消费者组的概念，组内每个消费者并行消费
   • 一个分区的数据，只能由消费者组内的一个消费者消费，避免重复消费问题
   • 如果没有对消费者分组，那么一个分区的数据可以被多个消费者消费
3. 为提高可用性，为每个partition增加若干副本，类似NameNode HA
   • 消费者只能从主partition进行消费，当主partition挂掉后，从partition可成为主partition
4. Zookeeper（Kafka2.8.0之前必须使用Zookeeper，2.8.0之后可选karft）
   • 记录集群中哪些broker服务器上线了
   • 记录哪个partition是主
     

5.下载安装

5.1下载

1. 官网地址：https://kafka.apache.org/downloads

   wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.12-3.0.0.tgz -P /opt/software
   

2. 解压

   tar -zxvf /opt/software/kafka_2.12-3.0.0.tgz -C /opt/moudle
   

3. 重命名

   mv /opt/moudle/kafka_2.12-3.0.0 /opt/moudle/kafka
   

5.2修改配置

1. 修改server.properties

# 集群中每个broker的身份标识，整个集群中不能存在重复的id
broker.id=0

# 数据存储目录，建议不要使用/tmp临时路径
# log.dirs=/tmp/kafka-logs
log.dirs=/opt/moudle/kafka/datas

# 连接的zookeeper集群
# /kafka为了方便管理数据
# zookeeper.connect=localhost:2181
zookeeper.connect=first-node:2181,second-node:2181,third-node:2181/kafka

5.3集群分发

1. 分发kafka软件包

   xsync /opt/moudle/kafka
   

2. 修改集群其他节配置文件的broker.id

   ssh second-node "sed -i.bak s/broker.id=0/broker.id=1/ /opt/moudle/kafka/config/server.properties"
   ssh third-node "sed -i.bak s/broker.id=0/broker.id=2/ /opt/moudle/kafka/config/server.properties"
   

3. 配置kafka环境变量：sudo vim /etc/profile.d/my_env.sh

   # KAFKA_HOME
   export KAFKA_HOME=/opt/moudle/kafka
   export PATH=$PATH:$KAFKA_HOME/bin
   

4. 重新加载环境变量

   source /etc/profile
   

5. 环境变量配置文件分发

   • 如果root用户未配置ssh免密登录，此处需要输入密码

   sudo /home/fatpuffer/bin/xsync /etc/profile.d/my_env.sh
   

6. 登录集群环境其他节点重新加载环境变量

   source /etc/profile
   

5.4启动kafka

5.4.1首先需要启动ZK集群

zk.sh start

5.4.2启动Kafka

kafka-server-start.sh -daemon /opt/moudle/kafka/config/server.properties

5.4.3停止Kafka

kafka-server-stop.sh

6.Kafka集群

6.1启动停止脚本

• 1.用户bin目录下创建kfk.sh文件：vim /home/fatpuffer/bin/kfk.sh

  #! /bin/bash
  
  case $1 in
  "start"){
      for node in first-node second-node third-node
      do
          echo ----------------------kafka $node start--------------------
          ssh $node "kafka-server-start.sh -daemon /opt/moudle/kafka/config/server.properties"
      done
  };;
  "stop"){
      for node in first-node second-node third-node
      do
          echo ----------------------kafka $node stop--------------------
          ssh $node "kafka-server-stop.sh"
      done
  };;
  esac
  

• 2.修改脚本权限

  comod +x kfk.sh
  

• 3.脚本使用

  kfk.sh start
  kfk.sh stop
  

6.2注意事项

1. 启动Kafka前一定要确保ZK集群已经启动成功
2. 必须等待Kafka完全停止，才可以停止Zoopeeper，否则会导致Kafka无法停止

6.3Kafka在ZK集群中的节点

1. 查看节点/kafka信息

   [zk: localhost:2181(CONNECTED) 1] ls /kafka
   [admin, brokers, cluster, config, consumers, controller_epoch, feature, isr_change_notification, latest_producer_id_block, log_dir_event_notification]
   

2. 查看节点/kafka/brokers

   [zk: localhost:2181(CONNECTED) 16] ls /kafka/brokers
   [ids, seqid, topics]
   

3. 查看节点/kafka/brokers/ids

   [zk: localhost:2181(CONNECTED) 11] ls /kafka/brokers/ids
   [0, 1, 2]
   

4. 查看节点/kafka/brokers/topics

   [zk: localhost:2181(CONNECTED) 17] ls /kafka/brokers/topics
   []
   

5. 获取/kafka/brokers/ids/0的数据

   [zk: localhost:2181(CONNECTED) 13] get /kafka/brokers/ids/0
   {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://first-node:9092"],"jmx_port":-1,"features":{},"host":"first-node","timestamp":"1670480025960","port":9092,"version":5}
   [zk: localhost:2181(CONNECTED) 14] get /kafka/brokers/ids/1
   {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://second-node:9092"],"jmx_port":-1,"features":{},"host":"second-node","timestamp":"1670480032519","port":9092,"version":5}
   [zk: localhost:2181(CONNECTED) 15] get /kafka/brokers/ids/2
   {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://third-node:9092"],"jmx_port":-1,"features":{},"host":"third-node","timestamp":"1670480038549","port":9092,"version":5}
   

7.Kafka基本命令行操做

7.1脚本图鉴

7.2主题命令操做

1. 命令参数

   参数	描述
   –bootstrap-server <String: server connect to>	连接的 Kafka Broker主机名称和端口
   –topic <String: topic>	操做的Topic名称
   –create	创建主题
   –delete	删除主题
   –alter	修改主题
   –list	查看所有主题
   –describe	查看主题详细描述
   –partitions <Integer: # of partitions>	设置分区数
   –replication-factor <Integer: replication factor>	设置分区副本
   –config <String: name=value>	更新系统默认的配置
   –producer-property	将自定义属性传递给生成器的机制，形如：key=value
   –producer.config	生产者配置属性文件 [–producer-property] 优先于此配置
   –property	自定义消息读取器：parse.key=true|false key.separator=<key.separator> ignore.error=true
   –request-required-acks	生产者请求的确认方式：0、1(默认值)、all
   –sync	同步发送消息

2. 查看当前服务器中的所有topic

   kafka-topics.sh --bootstrap-server first-node:9092 --list
   
   # 连接多个节点
   # kafka-topics.sh --bootstrap-server first-node:9092,second-node:9092,third-node:9092 --list
   

3. 创建一个名为demo1的主题，设置分区数为1，分区副本为3

   kafka-topics.sh --bootstrap-server first-node:9092 --topic demo1 --create --partitions 1 --replication-factor 3
   

   [zk: localhost:2181(CONNECTED) 22] ls /kafka/brokers/topics
   [demo1]
   [zk: localhost:2181(CONNECTED) 23] ls /kafka/brokers/topics/demo1
   [partitions]
   [zk: localhost:2181(CONNECTED) 24] ls /kafka/brokers/topics/demo1/partitions
   [0]
   [zk: localhost:2181(CONNECTED) 25] ls /kafka/brokers/topics/demo1/partitions/0
   [state]
   [zk: localhost:2181(CONNECTED) 26] ls /kafka/brokers/topics/demo1/partitions/0/state
   []
   
   [zk: localhost:2181(CONNECTED) 27] get /kafka/brokers/topics/demo1
   {"removing_replicas":{},"partitions":{"0":[2,1,0]},"topic_id":"D6Lse1d5TduQCu0mh-idwQ","adding_replicas":{},"version":3}
   
   [zk: localhost:2181(CONNECTED) 28] get /kafka/brokers/topics/demo1/partitions/0/state
   {"controller_epoch":5,"leader":2,"version":1,"leader_epoch":0,"isr":[2,1,0]}
   

4. 查看demo1主题详情

   kafka-topics.sh --bootstrap-server first-node:9092 --describe demo1 
                                                   # 分区数             # 分区副本数              # 分区大小
   Topic: demo1	TopicId: D6Lse1d5TduQCu0mh-idwQ	PartitionCount: 1	ReplicationFactor: 3	Configs: segment.bytes=1073741824
   	                # 分区开始位置    # 副本中的leader对应的broker   # 分区副本对应的broker（这三个副本分别存储在broker.id=2、broker.id=1、broker.id=0）  
   	Topic: demo1	Partition: 0	Leader: 2		            Replicas: 2,1,0 Isr: 2,1,0
   

   • 生产者和消费者只能针对leader节点进行操做

5. 修改主题demo1分区数为3

   kafka-topics.sh --bootstrap-server first-node:9092 --topic demo1 --alter --partitions 3
   

   • 注意：分区数只能增加，不能减少

   [zk: localhost:2181(CONNECTED) 34] ls /kafka/brokers/topics
   [demo1]
   [zk: localhost:2181(CONNECTED) 35] ls /kafka/brokers/topics/demo1
   [partitions]
   [zk: localhost:2181(CONNECTED) 36] ls /kafka/brokers/topics/demo1/partitions
   [0, 1, 2]
   [zk: localhost:2181(CONNECTED) 37] ls /kafka/brokers/topics/demo1/partitions/1
   [state]
   [zk: localhost:2181(CONNECTED) 38] ls /kafka/brokers/topics/demo1/partitions/1/state
   []
   
   [zk: localhost:2181(CONNECTED) 39] get /kafka/brokers/topics/demo1
   {"removing_replicas":{},"partitions":{"2":[1,2,0],"1":[0,1,2],"0":[2,1,0]},"topic_id":"D6Lse1d5TduQCu0mh-idwQ","adding_replicas":{},"version":3}
   
   [zk: localhost:2181(CONNECTED) 40] get /kafka/brokers/topics/demo1/partitions/0/state
   					   # 0号分区leader为broker2
   {"controller_epoch":5,"leader":2,"version":1,"leader_epoch":0,"isr":[2,1,0]}
   
   [zk: localhost:2181(CONNECTED) 41] get /kafka/brokers/topics/demo1/partitions/1/state
                          # 1号分区leader为broker0 
   {"controller_epoch":5,"leader":0,"version":1,"leader_epoch":0,"isr":[0,1,2]}
   
   [zk: localhost:2181(CONNECTED) 55] get /kafka/brokers/topics/demo1/partitions/2/state
   					   # 2号分区leader为broker1
   {"controller_epoch":5,"leader":1,"version":1,"leader_epoch":0,"isr":[1,2,0]}
   

   

   [fatpuffer@first-node ~]$ kafka-topics.sh --bootstrap-server first-node:9092 --describe demo1 
   Topic: demo1	TopicId: D6Lse1d5TduQCu0mh-idwQ	PartitionCount: 3	ReplicationFactor: 3	Configs: segment.bytes=1073741824
   	Topic: demo1	Partition: 0	Leader: 2	Replicas: 2,1,0	Isr: 2,1,0
   	Topic: demo1	Partition: 1	Leader: 0	Replicas: 0,1,2	Isr: 0,1,2
   	Topic: demo1	Partition: 2	Leader: 1	Replicas: 1,2,0	Isr: 1,2,0
   

6. 生产者向broker中的demo1主题发送一条消息

   kafka-console-producer.sh --bootstrap-server first-node:9092 --topic demo1
   

7. 消费者连接broker，并且监听deno1主题，broker监控到demo1主题中有消息产生，则将该消息推送给所有demo1主题的监听者

   kafka-console-consumer.sh --bootstrap-server first-node:9092 --topic demo1
   
   # --from-beginning可以获取历史数据
   kafka-console-consumer.sh --bootstrap-server first-node:9092 --topic demo1 --from-beginning
   

   

8. 发送带key的消息

   kafka-console-producer.sh --bootstrap-server first-node:9092 --topic demo1 --producer-property parse.key=true
   

   • 键值之间使用tab键分割

   

8.生产者（producer）

8.1发送原理

在消息发送过程中，涉及到了两个线程，main线程和sender线程。在main线程中创建了一个双端队列：RecordAccumulator。main线程将消息发送给：RecordAccumulator，sender线程不断从：RecordAccumulator中拉取消息发送到：Kafka Broker

8.2API发送流程

1. 创建配置对象

2. 向配置对象写入kafka集群连接信息

3. 向配置对象写入序列化信息

4. 根据配置对象，创建生产者对象

5. 使用生产者对象发送数据

6. 关闭生产者对象

   kafka-python：https://kafka-python.readthedocs.io/en/master/apidoc/KafkaConsumer.html

8.3生产者分区

8.3.1分区好处

1. 便于合理使用存储资源，每个Partion在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
2. 提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据

8.3.2生产者发送消息的分区策略

1. 在发送消息时指定了分区，则所有消息发送到该分区
2. 在发送消息时未指定分区，但指定了key
   • 将key的hash值与topic的partition数进行取余得到partition值
     • 例如：key1的hash值为5，key2的hash值为6，topic的partition数为2，那么key1对应的value1就写入1号分区，key2对应的value2写入0号分区
3. 在发送消息时未指定分区，也未指定key
   • Kafka采用Sticky Partition(黏性分区器)，会随机选择一个分区，并尽可能一直使用该分区，等待该分区的batch.size已满或已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）
     • 例如：第一次随机选择0号分区，等0号分区当前批次（batch.size）已经满了（默认16k），或者linger.ms设置的时间到了，Kafka再随机选择一个分区进行使用（如果是0，还会继续随机）
     • Java：hashcode------------------------->Python：ord("a")

8.3.3自定义分区器

• 根据发送数据内容进行过滤，将不同类型数据发送到不同分区
  

8.4提高吞吐量

• 1.主要涉及两个参数的调优，当linger.ms时常，越接近batch.size：16kb，则效率越高
  • 1.batch.size：传输量，16或32
  • 2.linger.ms：延迟，5-10ms
• 2.数据压缩
  • compression.type：压缩snappy
• 3.RecordAccumulator：缓冲区大小，修改为64M
  

8.5数据可靠性

8.5.1生产者发送过来数据就不管了：ack=0，可靠性差，效率高

8.5.2可靠性总结

1. acks=0，生产者发送过来数据就不管了，可靠性差，效率高
2. acks=1，生产者发送过来数据，Leader（落盘）应答，可靠性中等，效率中等
3. acks=-1，生产者发送过来数据，Leader和ISR队列里面的所有Follower应答，可靠性高，效率低
4. 生产环境中，acks=0很少使用；acks=1一般使用传输普通日志，允许丢个别；acks=-1一般用于和钱相关的数据，对可靠性要求比较高的场景。
   

8.5.2数据重复

1.产生原因

• acks=-1(all)：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐后应答
  • 重复原因分析：当Leader收到消息，且已经被其他ISR队列中的Follower同步完成，但是再应答生产者时，Leader挂了，Kafka重新选举出Leader。生产者未收到应答，会再次发送数据，新的Leader拿到数据后，其他ISR队列中的Follower又来拉取一次数据，此时数据就出现了重复。

2.数据传递语义

• 至少一次（At Least Once）：ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答得最小副本数大于等于2
• 最多一次（At Most Once）：ACK级别设置为0

3.总结

• At Least Once：可以保证数据不丢失，但是不能保证数据不重复
• At Most Once：可以保证数据不重复，但是不能保证数据不丢失

4.精确一次（Exact Once）

• 对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复，也不能丢失
• Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务

5.幂等性原理

• 幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
• 精确一次（Exact Once）= 幂等性 + 至少一次（At Least Once）（ack = -1 + 分区副本数 >= 2 + ISR最小副本数 >= 2）

6.重复判断的标准

• 具有<PID，Partition，SeqNumber>相同主键的消息提交时，Broker只会持久化一条。
  • PID是kafka每次重启都会分配一个新的；
  • Partition标识分区号；
  • Sequence Number是单调自增的。
• 所以幂等性只能保证的是在单分区单会话内不重复

7.开启幂等性

• enable.idempotence，默认为true，false为关闭

8.6事务

1.为什么要使用事务

• 幂等性只能保证单分区、单会话内数据不重复，一旦kafka挂掉，重启后PID就会变化，此时就无法保证数据不重复
• 如果分区号变化，也依然无法保证数据不重复

2.事务原理

1. 开启事务，必须开启幂等性
   

3.API操做

8.7.数据有序

8.8.数据乱序

1. kafka在1.x版本之前，保证数据单分区有序，条件如下：
   • max.in.flight.requests.per.connection=1(不需要考虑是否开启幂等性)
2. kafka在1.x版本之后，保证数据单分区有序，条件如下：
   • 1.未开启幂等性
     • max.in.flight.requests.per.connection=1
   • 2.开启幂等性
     • max.in.flight.requests.per.connection需要设置小于等于5
     • 原因说明：因为在kafka1.x之后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近的5个request的数据是有序的（根据序列号单调递增，如果有序，直接落盘，遇到无序的，先缓存内存中）

9.Broker

9.1.1ZK存储Kafka信息

9.2Broker工作流程

9.3节点服役退役

9.3.1服役新节点

1. 准备新节点
   • 1.基于历史节点克隆（克隆前关机克隆目标节点）
   • 2.修改IP地址
   • 3.修改主机名hostname
   • 4.重新启动克隆目标节点和克隆节点
   • 5.修改克隆后节点kafka配置文件中的broker.id
   • 6.删除克隆后节点kafka数据文件（datas和logs）
   • 7.启动原有集群
   • 8.单独启动新克隆节点的kafka
2. 执行负载均衡（历史主题等数据，不会出现在新节点上，所以需要使用负载均衡来迁移）
   • 1.创建一个要均衡的主题：vim topics-to-move.json

     {
     	"topics": [
     		"topic": "demo1"
     	],
     	"version": 1
     }
     

   • 2.生成一个负载均衡计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
     

   • 3.创建副本存储计划（所有副本存储在broker0、broker1、broker2、broker3中）：vim increase-replication-factor.json
     • 来源于上述命令生成结果：Proposed partition reassignment configuration

     {"version":1, "partitions":[{"topics":"demo1", "partition":0, "replicase":[3, 1], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":1, "replicase":[0, 2], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":2, "replicase":[1, 3], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":3, "replicase":[2, 0], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":4, "replicase":[3, 2], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":5, "replicase":[0, 3], "log_dirs":["any", "any"]}, {"topics":"demo1", "partition":6, "replicase":[1, 0], "log_dirs":["any", "any"]}]}
     

   • 4.执行副本存储计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --execute
     

     • 出现Successfully started...即表示执行成功
   • 5.验证副本存储计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --verify
     

9.3.1退役旧节点

1. 执行负载均衡操做（将要退役节点数据迁移到其他节点）

   • 1.创建一个要均衡的主题：vim topics-move.json

     {
     	"topics": [
     		"topic": "demo1"
     	],
     	"version": 1
     }
     

   • 2.创建执行计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate
     

   • 3.创建副本存储计划（所有副本存储在broker0、broker1、broker2中）：vim increase-replication-factor.json

     {"version":1, "partitions":[{"topics":"demo1", "partition":0, "replicase":[2, 0, 1], "log_dirs":["any", "any", "any"]}, {"topics":"demo1", "partition":1, "replicase":[0, 1, 2], "log_dirs":["any", "any", "any"]}, {"topics":"demo1", "partition":2, "replicase":[1, 2, 0], "log_dirs":["any", "any", "any"]}]}
     

   • 4.执行副本存储计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --execute
     

     • 出现Successfully started...即表示执行成功

   • 5.验证副本存储计划

     /opt/module/kafka/bin/kafka-reassign-partitions.sh --bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --verify
     

2. 停止退役后的kafka节点

   kafka-server-stop.sh
   

9.4kafka副本

9.4.1副本信息

1. Kafka副本作用：提高数据可靠性
2. Kafka默认1个副本，生产环境下一般配置2个，保证数据可靠性，太多副本会增加磁盘存储空间，增加网络传输时长，降低效率
3. Kafka副本分为Leader和Follower。Kafka生产者只会把数据发往Leader，然后Follower找Leader进行数据同步
4. Kafka分区中的所有副本系统称为AR(Assigned Repllicas)
   • AR = ISR + OSR
5. ISR：标识和Leader保持同步的Follwer集合，如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR队列。该时间的阈值由replica.lag.time.max.ms参数设定，默认30s。Leader发生故障后，就会从ISR中选出新的Leader
6. OSR：标识Follower和Leader副本同步时，延时过多的副本

9.4.2选举流程

• Kafka中有一个Broker的Controller会被选举为Controller Leader，负责管理集群broker的上下线，所有topic的分区副本分配和Leader选举等工作
• Controller的信息同步工作是依赖于Zookeeper的
• 1.启动Broker时，在zookeeper中/kafka/brokers/ids下进行注册
• 2.每个Broker中的Controller抢先注册zookeeper中的/kafka/controller节点，谁抢到了谁就是controller leader
• 3.controller leader监听/kafka/broker/ids的变化
• 4.Broker节点启动过半后，Controller开始选举Leader
  • 选举规则：在ISR中存活为前提，按照AR中排在前面的优先。例如：ar[1, 0 ,2]，isr[1, 0, 2]，那么leader就会按照1，0，2的顺序轮询
• 5.Controller选举出Leader后，开始向zookeeper的/kafka/brokers/topics/demo1/partitions/0/state节点写入信息
• 6.其他Broker上的Controller从zookeeper上同步相关信息，防止成为leader的Controller意外挂掉
• 7.假首次选举出的leader位于Broker1，此时如果Broker1上的Leader意外挂了
• 8.此时Controller监测到/kafka/brokers/ids上的变化
• 9.Controller从/kafka/brokers/topics/demo1/partitions/0/state节点获取ISR信息
• 10.再次按照选举规则，开始新一轮选举
• 11.选举完成后更新/kafka/brokers/topics/demo1/partitions/0/state节点信息（主要是leader，isr）

9.4.3手动调整分区副本存储

9.4.3.1原因

• 在生产环境中，每台服务器的配置和性能可能不一致，但是kafka只会根据自己的代码规则创建对应的分区副本，就会导致个别性能低的服务器存储压力较大，所以需要手动调整分区副本的存储
• 需求：创建一个新的topic，4个分区，2个副本，名称为demo2，将该topic的所有副本都存储到broker0和broker1两台服务器上
  

9.4.3.2流程

1. 创建demo2主题

   kafka-topics.sh -bootstrap-server first-node:9092 --create --topic demo2 --partitions 4 --replication-factor 2
   

2. 查看分区副本存储情况

   kafka-topics.sh -bootstrap-server first-node:9092 --describe --topic demo2 
   

   Topic: demo2	Topicld: -tfY5h@3Rg236b6sooxs7g	PartitionCount: 4	ReplicationFactor: 2	configs; segment. bytes=1073741824
   
   Topic: demo2	Partition: 0	Leader: 2	Replicas: 2,0	Isr: 2,0
   Topic: demo2	Partition: 1	Leader: 3	Replicas: 3,2	Isr: 3,2
   Topic: demo2	Partition: 2	Leader: 1	Replicas: 1,3	Isr: 1,3
   Topic: demo2	Partition: 3	Leader: 0	Replicas: 0,1	Isr: 0,1
   

3. 创建副本存储计划（将所有副本都指定存储在broker0和broker1中）：vim increase-replication-factor.json

   {
   	"version": 1,
   	"partitions": [
   		{"topic": "demo2", "partition": 0, "replicas": [0, 1]},
   		{"topic": "demo2", "partition": 1, "replicas": [0, 1]},
   		{"topic": "demo2", "partition": 2, "replicas": [0, 1]},
   		{"topic": "demo2", "partition": 3, "replicas": [0, 1]},
   	]
   }
   

4. 执行副本存储计划

   kafka-reassign-partitions.sh -bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --execute
   

5. 验证副本存储计划

   kafka-reassign-partitions.sh -bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --verify
   

6. 查看分区副本存储情况

   kafka-topics.sh -bootstrap-server first-node:9092 --describe --topic demo2 
   

9.4.4自动平衡（Leader Partition）

• 正常情况下，kafka本身会自动把Leader Partition平均分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机broker重启后都是follower partition，读写请求很低，造成负载不均衡。
• auto.leader.rebalance.enable：默认是true。自动平衡（如果机器性能不一致，且做了手动调整分区存储，那么不建议将其设置为true）
  • 频繁触发自动平衡，是需要消耗大量资源的
• leader.imbalance.per.broker.percentage：每个broker允许的不平衡leader的比率，默认10%，如果超过这个值，控制器会触发leader的平衡
• leader.imbalance.check.interval.seconds：检查leader负载是否平衡的间隔时间。
• 示例如下：
  
• 红色圈起来的部分，Leader和AR顺序排列不符，说明集群中有broker曾宕机过
• 针对broker0节点，分区2的AR优先副本是0节点，但是0节点却不是leader节点，所以不平衡数+1，AR副本总数是4
• 所以broker0节点的不平衡率为：1/4=25%，大于10%，会触发再平衡
• broker2和broker3节点和broker0节点不平衡率一样，也会触发再平衡，broker1的不平衡数为0，不需要再平衡

9.4.5增加副本因子

9.4.5.1原因

• 在生产环境中，由于某个主题的重要等级需要提升，我们考虑增加副本，副本数的增加需要先制定计划，然后根据计划执行

9.4.5.2流程

1. 创建topic

   kafka-topics.sh -bootstrap-server first-node:9092 --create --topic demo3 --partitions 3 --replication-factor 1
   

2. 手动增加副本存储，创建副本存储计划（所有副本都指定存储在broker0、broker1、broker2中）：vim increase-replication-factor.json

   	{
   		"version": 1,
   		"partitions": [
   			{"topic": "demo3", "partition": 0, "replicas": [0, 1, 2]},
   			{"topic": "demo3", "partition": 1, "replicas": [0, 1, 2]},
   			{"topic": "demo3", "partition": 2, "replicas": [0, 1, 2]}
   		]
   	}
   

3. 执行副本存储计划

   kafka-reassign-partitions.sh -bootstrap-server first-node:9092 --reassignment-json-file increase-replication-factor.json --execute
   

9.5文件存储

9.5.1文件存储机制

• Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应一个log文件，该log文件中存储的就是producer生产的数据。producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，kafka采取了分片和索引机制，将每个partition分为多个segment，每个segment包括：.index偏移量索引文件，.log日志文件，和.timeindex时间戳索引等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区号，例如：demo1-0。

• 消费者消费完成后不会删除数据，数据默认保存7天，由timeindex来计算

• 查看.log文件

  kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index Dumping ./00000000000000000000.index 
  

  Starting offset: 0
  baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: 0 isTransactional: false isControl: false position: 75 CreateTime：1670918886067 size: 75 magic: 2 compresscodec: none crc: 73416945 isvalid: true 
  

• 查看.index文件

  kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log Dumping ./00000000000000000000.log
  

  offset: 0 position: 0
  

9.5.1文件清除策略

• Kafka中默认的日志保存时间为7天，可以通过调整如下参数修改保存时间
  • log.retention.hours，最低优先级小时，默认7天
  • log.retention.minutes，分钟
  • log.retention.ms，最高优先级毫秒
  • log.retention.check.interval.ms，负责设置检查周期，默认5分钟
• 那么日志一旦超过了设置时间，怎么处理
  • Kafka中提供的日志清理策略有delete和compact两种
• delete：日志删除，将过期数据删除
  • log.cleanup.policy=delete：所有数据启用删除策略
  • 思考：如果一个segment中有一部分数据过期，一部分数据没有过期，怎么处理？
    • 基于时间：默认打开，以segment中所有记录中的最大时间戳作为该文件时间戳，如果最大时间戳距今超过7天，就直接删除
    • 基于大小：默认关闭，超过设置的所有日志总大小 ，删除最早的segment
      • log.retention.bytes默认等于-1，标识无穷大
• compact：日志压缩（对于相同的key的不同value值，只保留最新的，类似于python字典中的update方法）
  • 压缩后offset可能是不连续的，当从这些offset中消费消息时，将会拿到比这个offset大的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。
  • 这种策略只适合特殊场景，比如消息的key是用户id，value是用户资料，通过这种压缩策略，整个消息集里就保存了用户最新的资料

9.6高效读写数据

1. Kafka本身是分布式集群，可以采用分区技术，并行度高
2. 读数据采用稀疏索引，可以快速定位药消费的数据
3. 顺序写磁盘
   • kafka的producer生产数据，要写入log文件中，写的过程是一直追加到文件末端，为顺序写。官方测试结果：顺序写能达到600M/s，而随机写只有100K/s。这与磁盘的机械结构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间
4. 页缓存 + 零拷贝
   • 零拷贝：kafka的数据加工处理操做交由kafka生产者和kafka消费者处理，kafka broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。
   • 页缓存（PageCache）：kafka重度依赖底层操做系统提供的PageCache功能，当上层有写操作时，操做系统只是将数据写入PageCache。当度操做发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当作了磁盘使用

10.消费者（consumer）

10.1Kafka消费方式

1. pull（拉）模式：kafka采用
   • kafka采用从broker中主动拉取数据
   • pull模式的不足之处：如果broker没有数据，消费者可能会陷入循环中，一直返回空
2. push（推）模式：
   • kafka没有采取这种方式，因为broker决定消息发送速率，很难适应所有消费者的消费速率，例如推送的速度是：50m/s，而Consumer的消费能力如果只有20m/s，则来不及处理

10.2Kafka消费者工作流程

10.2.1消费者总体流程

1. 生产者生产消息，发往集群中每个broker上topic的leader分区，follower分区同步数据
2. 消费者从topic上的leader分区拉取数据，进行消费
   • 每个消费者可以从多个分区拉取数据，也可以多个消费者从同一个分区消费数据
   • 如果消费者建立分组，那么分组内的多个消费者不可以消费一个分区的数据
3. 如果一个消费者消费demo1分区中的数据，消费前该消费者挂了，此时其他消费者怎么知道demo1分区中的数据消费到哪里了？然后从未消费的地方开始消费，避免再次从头开始，出现重复消费
   • 新版的kafka有一个默认主题：__consumer_offsets，来记录主题内消费偏移信息，即使某一个消费者挂了，其他消费者依然可以从未消费的位置开始消费

10.2.2消费者组原理

• Consumer Group（CG）：消费者组由多个consumer组成，形成一个消费者组的条件，是所有消费者的gropuid相同
  • 消费者组内的每个消费者负责不同分区的数据，一个分区只能由组内的一个消费者消费
  • 消费者组之间互不影响。所有的消费者都属于某个消费者组。即消费者组是逻辑上的一个订阅者

10.2.3消费者组初始化流程

1. coordinator：辅助实现消费者组的初始化和分区的分配。
   • coordinator节点的选择 = groupid的hashcode值%50
     • 为什么是50， __consumer_offsets的分区数量 = 50
   • 例如：groupid的hashcode值=1，1%50=1，那么__consumer_offsets主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这点消费者组的老大，消费者组下的所有消费者提交offset的时候就往这个分区去提交offset

10.2.4消费者组详细消费流程

10.3Kafka消费者API

1. 独立消费者（订阅主题）
   
   
2. 独立消费者（订阅分区）
   
3. 消费者组
   • 将独立消费者（订阅主题）代码启动三份即可

10.4Kafka分区的分配以及再平衡

10.4.1分区平衡

• 一个consumer group中有多个consumer组成，一个topic有多个partition组成，现在的问题是，到底是由哪个consumer来消费哪个partition的数据

10.4.2分区平衡策略

1. Range
2. RoundRobin
3. Sticky
4. CooperativeSticky
5. 可以通过配置参数：partition.assignment.stratge，修改分区的配置策略。默认策略是：Range + CooperativeSticky，kafka可以同时使用多个分区配置策略。

10.4.3Range分区策略

1. Range是对每个topic而言
2. 首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序排序
   • 假设现在有7个分区，3个消费者，排序后的分区将会是：0，1，2，3，4，5，6，消费者排序后：C0，C1，C2
3. 通过partitions数 / consumer数来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者会多消费一个分区。
   • 例如，7 / 3 = 2 余 1，那么消费者C0便会多消费一个分区。8 / 3 = 2 余 2，那么C0，C1分别多消费一个分区
4. 注意：如果只是针对一个topic而言，C0消费者多消费一个分区影响不是很大，但是如果有N多个topic，那么针对每个topic消费者C0都将多消费一个分区，topic越多，C0消费的分区会比其他消费者明显多消费N个分区，容易产生数据倾斜。
5. 如果C0消费者挂了，生产者在45s内，发送过来的消息，将全部直接交由存活的消费者中的某一个，而不是均分
   • C1：0、1、2、3、4 --------------> C2：4、5
   • 或者
   • C1：3、4 --------------> C2：0、1、2、4、5
6. 超过45s该消费者还未修复的话，就会触发再平衡，此时按照剩余的消费者进行range策略
   • C1：0、1、2、3
   • C2：4、5、6
     

10.4.4RoundRobin分区策略

1. RoundRobin：针对集群中所有topic而言
2. 把所有的partition和所有的consumer都列出来，然后按照hashcode进行排序，最后通过轮询算法来分配partition给到各个消费者
3. 假使C0这个消费者挂了，生产者在45s内，向C0所负责消费的分区0、3、6，发送过来的消息，将全部直接交由其他存活的消费者（C1、C2），0交由C1、3交由C2、6交由C1
   • C1：0、1、4、6分区
   • C2：2、3、5分区
4. 超过45s该消费者还未修复的话，就会触发再平衡，此时按照剩余的消费者进行range策略
   • C1：0、2、4、6
   • C2：1、3、5
     
5. 设置分区分配策略
   

10.4.5Sticky以及再平衡

1. 黏性分区定义：分配的结果带有黏性的，即再执行一次新的分配前，考虑上一次分配的结果，尽量少的调整分配的变动，可以减少大量的开销
2. 黏性分区是从kafka0.11版本开始引入的，首先会尽量均衡的放置分区到消费者上，在出现同一消费者组内消费者出现问题的时候，会保持原有分配的分区不会变化
3. 消费者随机均分分区
4. 假使C1这个消费者挂了，生产者在45s内，发送过来的消息，将全部直接交由其他存活的消费者
   • 假使C1挂之前分区如下：
     • C0：0、1分区
     • C1：4、5、6分区
     • C2：2、3分区
   • C1分区数据超时后交给C0和C2随机均分
     • C0：0、1、4、6分区
     • C2：2、3、5分区
5. 超过45s该消费者C1还未修复的话，就会触发再平衡，此时按照剩余的消费者进行range策略
   • C1：0、1、4、6分区
   • C2：2、3、5分区
6. 设置分区分配策略
   

10.5offset位移

10.5.1概念

• 从0.9版本开始，consumer默认将offset保存在kafka一个内置的topic中，该topic为__consumer_offsets
• 0.9版本之前，consumer默认将offset保存在zookeeper中（所有消费者都要和zookeeper交互，效率低下）
• __consumer_offsets主题里面采用了key和value的方式存储数据，key：group.id + topic + 分区号，value就是当前offset的值。每隔一段时间，kafka内部会对这个topic进行compact（压缩：覆盖更新），也就是每个roup.id + topic + 分区号就保留最新数据

10.5.2查看__consumer_offsets这个系统主题

1. 在配置文件/opt/module/kafka/config/consumer.properties中添加配置exclude.internal.topics=false，默认为true，表示不能消费系统主题，为了查看系统主题数据，所以需要修改为false
2. xsync分发脚本
3. 查看__consumer_offsets主题数据

   kafka-console-consumer.sh --bootstrap-server first-node:9092 --topic __consumer_offsets --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning
   

10.5.3自动提交

• 为使我们能够能够专注于自己的业务逻辑，kafka提供了自动提交offset的功能
• 自动提交offset相关参数
  • enable.auto.commit：是否开启自动提交offset功能，默认是true
  • auto.commit.interval.ms：自动提交offset的间隔时间，默认是5s
    

10.5.3手动提交

• 虽然自动提交offset十分便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机，因此kafka还提供了手动提交offset的api
• 手动提交offset的方法有两种，分别是：commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败尝试（由不可控因素导致，也会出现提交失败）；而异步提交没有失败重试机制，故有可能提交失败。
  • commitSync：必须等待offset提交完毕，再去消费下一批数据。
  • commitAsync：发送完提交offset请求后，就开始消费下一批数据
    

10.5.4指定offset消费

• 当kafka中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？
• auto.offset.reset=earliest|latest|none默认是latest
  • earliest：自动将偏移量重置为最早的偏移量，--from-beginning
  • least：自动将偏移量重置为最新偏移量
  • none：如果未找到消费者组的先前偏移量，则向消费者抛出异常
    

10.5.4指定时间消费

• 需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据
  

10.5.4漏消费和重复消费

• 重复消费：已经消费了的数据（自动提交offset）
  

• 漏消费：先提交offset后消费，有可能会造成数据的漏消费（手动提交offset）

  • 当消费者从分区中拿到要消费数据后，会立即提交offset，数据还存在内存中，未落盘，此时消费者如果挂掉，那么offset已经提交，单数数据未处理，导致这部分内存中的数据丢失。

• 解决方法：消费者事务

10.6消费者事务

10.6.1条件

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

10.7消费者数据积压

10.7.1消费者如何提高吞吐量

• 如果kafka消费能力不足，则可以考虑增Topic的分区数，并且同时提升消费组的消费者数量，消费者数=分区数。（两者缺一不可）
• 如果是下游数据处理不及时：提高每批次拉取数据的数量。批次拉取数量过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压