kafka基础-文末思维导图

**文末尾有思维导图**，文字就是思维导图的内容，如果不想看着，**可以直接拉到末尾，查看思维导图！**

注： 文章，是我学习了极客时间的《Kafka核心技术与实战》专栏总结的学习笔记。

# kafka基础

## 核心术语

1. Topic 主题

2. Partition 分区，一个主题多个分区

3. Record消息

4. 副本Replica，为消息提供冗余

4.1 leader副本，对外提供服务

4.2 follower副本，仅作为冗余数据

5. 消息位移Offset： 分区中每条消息的位置，单调递增 

### Producer生产者

### Consummer消费者

#### 消费者位移：记录消费者的进度，每个消费者都有自己的位移

#### 消费者组：同一个消费组下，同一个Topic下，一个分区，有且仅有一个消费者消费

#### 消费者组重平衡：一个消费组内有消费者挂了，其他消费者自动重分主题分区的过程。消费者端高可用手段。

## Broke

### 集群规划注意事项：

|因素|考量点|建议|

|--|--|--|

|操作系统|操作系统/IO模型|将kafka部署在Linux上，利用epoll模型|

|磁盘|IO性能|普通机械磁盘，kafka副本+分区机制，可以不考虑搭建RAID|

|磁盘容量|消息数，留存时间，平均消息大小，备份数估算磁盘容量|建议预留20%-30%|

|带宽|根据实现带宽资源与业务SLA估算服务器的数量|千兆带宽，建议每台服务器按照700Mbps来计算，避免大流量下的丢包|

### 4步集群磁盘规划

1. 每日需要的磁盘净容量（GB）= 每条消息平均大小（KB）*每日消息数*副本数 /1000/1000

2. 考虑索引等数据每日磁盘容量（GB）=每日需要的磁盘容量* 1.1

3. 不考虑压缩的磁盘总大小（GB）=考虑索引等数据每日磁盘容量 * 留存时间

4. 考虑压缩的磁盘总大小（GB）=不考虑压缩的磁盘总大小*0.75

### 参数配置

#### Broker重要参数

##### 与存储有关

###### log.dir和log.dirs

1. 建议log.dirs按逗号分割,

2. 目录挂在在多个物理磁盘上。提升读写与故障恢复

##### 与Zookeeper相关

###### zookeeper.connect 按逗号分割，记录Zookeeper集群的地址

##### 与Broker连接相关

###### listener,advertised.liteners 格式为：<协议名称，主机名，端口号>   

##### Topic管理相关

###### auto.create.topic.enable 建议fasle，是否自动创建主题

###### unclean.leader.election.enable 建议false，不允许非ISR副本，提升为leade

###### auto.leader.rebalance.enable 是否自动换leader ，建议false

##### 数据留存

###### broker级别

1. log.retention.{hours|minutes|ms} 一条消息保存多长时间

2. 优先级ms>minutes>hours

3. log.retention.bytes： 保存消息的总容量大小，默认-1 不限制

4. message.max.bytes 单条消息最大字节，默认1000012 不足1MB，建议设置大些

###### Topic级别参数限制

1. retention.ms规定该Topic消息被保存的时长

2. retention.bytes 规定了要为该Topic 预留多大的磁盘空间

3. max.message.bytes 决定kafka Broker能够正常接受该Topic的最大消息大小

##### JVM参数

###### KAFKA_HEAP_OPS: 指定堆大小

推荐：KAFKA_HEAP_OPTS=--Xms6g  --Xmx6g

###### KAFKA_JVM_PERFORMANCE_OPTS: 指定GC参数

首选G1，次选CMS

```

-server   按照server模式

-XX:+UseG1GC    使用G1回收器

-XX:MaxGCPauseMillis=20   表示每次GC最大的停顿毫秒数20ms

-XX:InitiatingHeapOccupancyPercent=35   当整个堆占用超过某个百分比时，就会触发并发GC周期

-XX:+ExplicitGCInvokesConcurrent   显式的对 GC 的触发也是并发执行

-Djava.awt.headless=true  java.awt.headless是J2SE的一种模式，用于在缺失显示屏、鼠标或者键盘时的系统配置。对于后端服务来讲，很多都是需要将这个属性设置为true的

```

#### 操作系统配置

##### 文件描述符限制 ulimit -n 1000000

##### 文件系统类型 XFS 的性能要强于 ext4

##### Swappiness 一个比较小的值。当使用swap时，可以观察到Broker 性能急剧下降

##### Flush 落盘时间 默认是 5 秒 。kafka有分区+副本机制，可以适当调大

## 生产者

### 分区

#### 每条消息，只会保存在某个分区中

#### 分区是负载均衡以及高吞吐量的关键

#### Kafka 分区策略

##### 默认分区策略：指定了 Key，使用消息键保序策略；没指定 Key，使用轮询策略。

##### 其他常见分区策略：常见的，轮询策略，随机策略，按消息键保序策略，按地理位置分区策略 

### 压缩算法

#### Producer端压缩、Broker端保存、Consumer端解压

#### Broker端重新压缩消息的2种情况

##### Broker端压缩算法与Producer端压缩算法不同

##### 兼容老版本格式的转换

#### 压缩算法

##### 吞吐量方面：LZ4>Snappy>zstd,GZIP

##### 压缩比率： zstd>LZ4>GZIP>Snappy

#### 启动压缩的条件

##### Producer运行机器本身CPU充足

##### 带宽资源有限

##### 千兆网络,CPU资源充足，建议开启zstd 

### 如何管理TCP连接

#### Kafka社区采用TCP作为底层通讯协议

#### 在创建KafkaProducer实例时创建TCP连接

##### 创建时机

###### 发送消息时

###### 更新元数据后

##### 谁负责连接

###### 创建KafkaProducer实例时，生产者应用会在后台创建一个Sender的线程，该线程会与Broker进行连接

##### 会连接谁

###### Producer会对所有bootstrap.servers指定的Broker进行连接，生产环境中，建议指定3-4台broke

#### 关闭TCP

##### 用户主动关闭（kill -9）

##### kafka自动关闭(connections.max.idle.ms=-1 关闭，默认是9分钟)  

## 消费者

### 消费者组

#### 提供的可扩展且具有容错性的消费者机制

#### 传统模型的实现

##### 所有实例都属于同一个Group，就实现了消息队列模型

##### 所有实例分属不同的Group，就实现了发布订阅模型

#### 特性

##### Consumer Group下有一个或多个Consumer实例

##### Group ID标示唯一的一个Consumer Group

##### Consumer Group下所有实例订阅主题的单个分区，只能分配给组内的某个Consumer实例消费。

### 位移

#### 位移主题

##### __consumer_offsets保存Kafka消费者的位移

#### 消息格式

##### 消息Key 

###### 保存 3 部分内容：<Group ID，主题名，分区号 >

##### 消息体

###### 消息体1： 位移值+元数据  

###### 消息体2：保存Consumer Group的消息，用来注册Consumer Group   

###### 消息体3：删除Group过期位移，或删除Group的消息。tombstone消息，delete mark，特点是消息体为null

#### 何时创建主题

##### 第一个Consumer程序启动时，Kafka会自动创建位移主题，默认分区50，副本数是3

####  Kafka使用Compact（压实）策略

##### 作用：删除位移主题中的过期消息，避免该主题无限期膨胀

##### 过程：Compact的过程就是扫描日志的所有消息，剔除哪些过期的消息，把剩下的消息整理在一起。

##### 什么是过期消息：同一个Key两条消息M1，M2，若M1的发送时间早于M2，那么M1就是过期消息 。

### 位移提交

#### 自动提交

##### enable.auto.commit设置为true，默认为true

#### 手动提交

##### enable.auto.commit设置为false

##### 提交方式

###### 同步位移提交：调用API，KafkaConsumer#commitSync().

###### 异步提交位移：调用KafkaConsumer#commitAsync().

###### 精细化位移管理

1. 同步：commitSync(Map<TopicPartition，OffsetAndMetadata>)

2. 异步：commitAsync(Map<TopicPartition,OffsetAndMetadata>);

#### CommitFailedException 异常处理

##### 常见产生原因

###### 消息处理时间超过了max.poll.interval.ms

##### 如何预防

###### 缩短单条消息处理时间

###### 增加Consumer端允许下游消费一批消息的最大时长

###### 减少下游系统，一次性消费的消息总数

###### 下游系统使用多线程来加速消费 

### 多线程消费者

#### 多线程+多KafkaConsumer实例

##### 优点：方便，速度快，分区内消费顺序易维护

##### 缺点：系统资源占用多，受限于分区数，扩展性差，线程自己处理消息容易超时从而引发Rebalance

#### 单KafkaConsumer+消息处理Worker线程池

##### 优点：扩展性好，伸缩性好

##### 缺点：实现难度高，难以维护分区内的消息消费顺序，处理链路长，不易位移提交管理

### 关联TCP连接

#### 3个时机

##### 发起FindCoordinator请求

##### 连接协调者时

##### 消费数据时

#### 3种连接

##### 确定协调者和获取集群元数据

##### 连接协调者，令其执行组成员管理操作

##### 执行实际的消息获取。  

### 监控消费进度

#### Kafka自带的命令行工具，Kafka-consumer-groups脚本。

#### Kafka Java Consumer API编程

#### 使用Kafka自带的JMX监控指标

##### records-lag-max

##### records-lead-min 消费者最小消费消息的位移与分区当前第一条消息位移的差值。

## 控制器

### 职责

#### 主题管理

#### 分区重分配

#### Preferred领导选举

#### 集群成员管理

#### 数据服务

### 重度依赖于Zookeepe

#### Zookeeper 概述

##### 高可用分布式协调服务框架

##### 类似于文件系统的树形结构，以"/"开头

##### znode分为持久和临时，临时的znode会话结束会删除

##### zonde发送变化，通过Watch通知功能

##### zookeeper，常用于集群成员管理，分布式锁，领导者选举

### 保存的重要数据

#### 所有Broker信息

#### 所有涉及运维任务的分区

### 选举规则

#### 第一个成功创建/controller节点的Broker会被指定为控制器。

### 注意事项

#### 集群工作环境中，控制器只能有一个

#### JMX的指标，activeController，监控有几个存活的控制器

### 0.11的改进  将多线程，改成了多线程加队列

## Kafka重要版本

### 0.11.0.0 提供幂等生产者，与事务API

### 1.0,2.0 kafka的streams的各种改进