kafka基础-文末思维导图kafka基础

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

kafka基础

核心术语

1. Topic 主题
2. Partition 分区，一个主题多个分区
3. Record消息
4. 副本Replica，为消息提供冗余 4.1 leader副本，对外提供服务 4.2 follower副本，仅作为冗余数据
5. 消息位移Offset： 分区中每条消息的位置，单调递增

Producer生产者

Consummer消费者

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

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

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

Broker

集群规划注意事项：

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副本，提升为leader

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台broker

关闭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领导选举

集群成员管理

数据服务

重度依赖于Zookeeper

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的各种改进