Kafka 基本原理.md

1. 初识kafka

1.1 kafka背景

Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写.

1.2 Kafka的使用场景

1. 日志收集：可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种消费端
2. 消息系统：解耦和生产者和消费者、缓存消息等。
3. 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这
   些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析。
4. 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反
   馈，比如报警和报告。

1.3 Kafka基本概念

从一个较高的层面上来看，producer通过网络发送消息到Kafka集群，然后consumer来进行消费，如下图：

服务端(brokers)和客户端(producer、consumer)之间通信通过TCP协议来完成

1.4 kafka 搭建以及使用

1.4.1 安装前的环境准备

由于Kafka是用Scala语言开发的，运行在JVM上，因此在安装Kafka之前需要先安装JDK

yum install java‐1.8.0‐openjdk* ‐y

java -version

kafka依赖zookeeper，所以需要先安装zookeeper

wget https://dlcdn.apache.org/zookeeper/zookeeper-3.5.10/apache-zookeeper-3.5.10-bin.tar.gz

tar ‐zxvf apache‐zookeeper‐3.5.8‐bin.tar.gz

mv apache-zookeeper-3.5.10-bin  zookeeper3.5.10

cp conf/zoo_sample.cfg conf/zoo.cfg

# 启动zookeeper
bin/zkServer.sh start
bin/zkCli.sh
#查看zk的根目录相关节点
ls / 
第一

#####1.4.2 下载kafka安装包
下载2.4.1 release版本，并解压：

# 下载解压
wget  https://archive.apache.org/dist/kafka/2.6.0/kafka_2.12-2.6.0.tgz



tar ‐xzf kafka_2.12-2.6.0.tgz

cd  kafka_2.12-2.6.0


修改配置

vim config/server.properties

config/server.properties

#broker.id属性在kafka集群中必须要是唯一
broker.id=0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://内网:9092               
advertised.host.name=外网ip或者域名
advertised.listeners=PLAINTEXT://外网ip或者域名:9092    


#kafka的消息存储文件
log.dir=/usr/local/data/kafka‐logs
#kafka连接zookeeper的地址
zookeeper.connect=139.198.187.102:2181

1.4.3 第三步：启动服务

现在来启动kafka服务：
启动脚本语法：kafkaserverstart.sh daemon server.properties
可以看到，server.properties的配置路径是一个强制的参数，daemon表示以后台进程运行，否则ssh客户端退出后，就会停止服务。(注意，在启动kafka时会使用linux主机名关联的ip地址，所以需要把主机名和linux的ip映射配置到本地host里，用vim /etc/hosts)

# 启动kafka，运行日志在logs目录的server.log文件里
./bin/kafka-server-start.sh -daemon ./config/server.properties config/server.properties #后台启动，不会打印日志到控制台或者用
bin/kafka-server-start.sh config/server.properties&
# 我们进入zookeeper目录通过zookeeper客户端查看下zookeeper的目录树
bin/zkCli.sh
ls / #查看zk的根目录kafka相关节点
ls /brokers/ids #查看kafka节点

# 停止kafka
 bin/kafka-server-stop.sh

1.4.4 server.properties核心配置详解

1.4.5 创建主题

现在我们来创建一个名字为“test”的Topic，这个topic只有一个partition，并且备份因子也设置为1：

bin/kafka-topics.sh --create --zookeeper 139.198.187.102:2181 --replication-factor 1 --partitions 1 --topic test


# 查看kafka中目前存在的topic

bin/kafka-topics.sh --list --zookeeper 139.198.187.102:2181

# 删除主题

bin/kafka-topics.sh --delete --topic test --zookeeper 139.198.187.102:2181


# 发送消息
bin/kafka-console-producer.sh --broker-list 139.198.187.102:9092 --topic test123

# 接收消息
bin/kafka-console-consumer.sh --bootstrap-server 139.198.187.102:9092 --topic test123

# 消费多主题
bin/kafka-console-consumer.sh --bootstrap-server 139.198.187.102:9092 --whitelist "test|test123"

# 单播消费
一条消息只能被某一个消费者消费的模式，类似queue模式，只需让所有消费者在同一个消费组里即可
分别在两个客户端执行如下消费命令，然后往主题里发送消息，结果只有一个客户端能收到消息

bin/kafka-console-consumer.sh --bootstrap-server 139.198.187.102:9092 --consumer-property group.id=testGroup --topic test123

#多播消费
一条消息能被多个消费者消费的模式，类似publish-subscribe模式费，针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者，该消费者属于testGroup-2消费组，结果两个客户端都能收到消息


bin/kafka-console-consumer.sh --bootstrap-server 139.198.187.102:9092 --consumer-property group.id=testGroup-2 --topic test123

# 查看消费组的消费偏移量
bin/kafka-consumer-groups.sh --bootstrap-server 139.198.187.102:9092 --describe --group testGroup



GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                               HOST             CLIENT-ID
testGroup       test123         0          24              24              0               consumer-testGroup-1-3ebfdde6-a623-4ed2-99d8-76a9e144c968 /139.198.187.102 consumer-testGroup-1


current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数



# 进入
cd /usr/local/data/kafka-logs

1.4.6 集群消费

log的partitions分布在kafka集群中不同的broker上，每个broker可以请求备份其他broker上partition上的数据。kafka
集群支持配置一个partition备份的数量。针对每个partition，都有一个broker起到“leader”的作用，0个或多个其他的broker作为“follwers”的作用。leader处理所有的针对这个partition的读写请求，而followers被动复制leader的结果，不提供读写(主要是为了保证多
副本数据与消费的一致性)。如果这个leader失效了，其中的一个follower将会自动的变成新的leader。

1.4.7 Producers

生产者将消息发送到topic中去，同时负责选择将message发送到topic的哪一个partition中。通过roundrobin
做简单的
负载均衡。也可以根据消息中的某一个关键字来进行区分。通常第二种方式使用的更多。

1.4.8 Consumers

传统的消息传递模式有2种：队列( queue) 和（publish-subscribe）
queue模式：多个consumer从服务器中读取数据，消息只会到达一个consumer。
publish-subscribe模式：消息会被广播给所有的consumer。
Kafka基于这2种模式提供了一种consumer的抽象概念：consumer group。
queue模式：所有的consumer都位于同一个consumer group 下。
publish-subscribe模式：所有的consumer都有着自己唯一的consumer group

如图所示：

1.4.9 消费顺序

一个partition同一个时刻在一个consumer group中只能有一个consumer instance在消费，从而保证消费顺序。
consumer group中的consumer instance的数量不能比一个Topic中的partition的数量多，否则，多出来的
consumer消费不到消息。
Kafka只在partition的范围内保证消息消费的局部顺序性，不能在同一个topic中的多个partition中保证总的消费顺序
性。

如果有在总体上保证消费顺序的需求，那么我们可以通过将topic的partition数量设置为1，将consumer group中的
consumer instance数量也设置为1，但是这样会影响性能，所以kafka的顺序消费很少用。

auto.offset.reset关乎kafka数据的读取，是一个非常重要的设置。常用的二个值是latest和earliest，默认是latest。

earliest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费。
latest ：当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费该分区下新产生的数据。


# 命令行参数

--to-earliest：把位移调整到当前最早位移处

--to-latest：把位移调整到当前最新位移处

--to-current：把位移调整到分区当前最新提交位移处

--to-offset <offset>： 把位移调整到指定位移处

--shift-by N： 把位移调整到当前位移 + N处，注意N可以是负数，表示向前移动

--to-datetime <datetime>：把位移调整到大于给定时间的最早位移处

相关api 配置参见官方文档：https://kafka.apache.org/21/documentation.html#consumerapi

2.Kafka设计原理详解

2.1Kafka核心总控制器Controller

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（KafkaController），它负责管理整个集群中所有分区和副本的状态。

• 当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
• 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
• 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

2.2Controller选举机制

在kafka集群启动的时候，会自动选举一台broker作为controller来管理整个集群，选举的过程是集群中每个broker都会尝试在zookeeper上创建一个 /controller 临时节点，zookeeper会保证有且仅有一个broker能创建成功，这个broker就会成为集群的总控器controller。
当这个controller角色的broker宕机了，此时zookeeper临时节点会消失，集群里其他broker会一直监听这个临时节点，发现临时节点消失了，就竞争再次创建临时节点，就是选举机制，zookeeper又会保证有一个broker成为新的controller。具备控制器身份的broker需要比其他普通的broker多一份职责，具体细节如下：

• 监听broker相关的变化。为Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。
• 监听topic相关的变化。为Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；为Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。
• 从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对于所有topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。
• 更新集群的元数据信息，同步到其他普通的broker节点中。

2.3 Partition副本选举Leader机制

controller感知到分区leader所在的broker挂了(controller监听了很多zk节点可以感知到broker存活)，controller会从ISR列表(参数unclean.leader.election.enable=false的前提下)里挑第一个broker作为leader(第一个broker最先放进ISR列表，可能是同步数据最多的副本)，如果参数unclean.leader.election.enable为true，代表在ISR列表里所有副本都挂了的时候可以在ISR列表以外的副本中选leader，这种设置，可以提高可用性，但是选出的新leader有可能数据少很多。

副本进入ISR列表有两个条件：

• 副本节点不能产生分区，必须能与zookeeper保持会话以及跟leader副本网络连通
• 副本能复制leader上的所有写操作，并且不能落后太多。(与leader副本同步滞后的副本，是由 replica.lag.time.max.ms 配置决定的，超过这个时间都没有跟leader同步过的一次的副本会被移出ISR列表)

2.4消费者消费消息的offset记录机制

每个consumer会定期将自己消费分区的offset提交给kafka内部topic：__consumer_offsets，提交过去的时候，key是consumerGroupId+topic+分区号，value就是当前offset的值，kafka会定期清理topic里的消息，最后就保留最新的那条数据
因为__consumer_offsets可能会接收高并发的请求，kafka默认给其分配50个分区(可以通过offsets.topic.num.partitions设置)，这样可以通过加机器的方式抗大并发。
通过如下公式可以选出consumer消费的offset要提交到__consumer_offsets的哪个分区
公式：hash(consumerGroupId) % __consumer_offsets主题的分区数

2.5消费者Rebalance机制

rebalance就是说如果消费组里的消费者数量有变化或消费的分区数有变化，kafka会重新分配消费者消费分区的关系。比如consumer group中某个消费者挂了，此时会自动把分配给他的分区交给其他的消费者，如果他又重启了，那么又会把一些分区重新交还给他。
注意：rebalance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebanlance。

如下情况可能会触发消费者rebalance

• 消费组里的consumer增加或减少了
• 动态给topic增加了分区
• 消费组订阅了更多的topic
  rebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS会有影响，如果kafka集群内节点较多，比如数百个，那重平衡可能会耗时极多，所以应尽量避免在系统高峰期的重平衡发生。

2.6消费者Rebalance分区分配策略：

主要有三种rebalance的策略：range、round-robin、sticky。
Kafka 提供了消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。默认情况为range分配策略。
假设一个主题有10个分区(0-9)，现在有三个consumer消费：

range策略就是按照分区序号排序，假设 n＝分区数／消费者数量 = 3， m＝分区数%消费者数量 = 1，那么前 m 个消费者每个分配 n+1 个分区，后面的（消费者数量－m ）个消费者每个分配 n 个分区。
比如分区 0~3 给一个consumer，分区 4~6 给一个consumer，分区 7~9 给一个consumer。

round-robin策略就是轮询分配，比如分区0、3、6、9给一个consumer，分区1、4、7给一个consumer，分区2、5、8给一个consumer
sticky策略初始时分配策略与round-robin类似，但是在rebalance的时候，需要保证如下两个原则。

1）分区的分配要尽可能均匀 。
2）分区的分配尽可能与上次分配的保持相同。
当两者发生冲突时，第一个目标优先于第二个目标 。这样可以最大程度维持原来的分区分配的策略。
比如对于第一种range情况的分配，如果第三个consumer挂了，那么重新用sticky策略分配的结果如下：
consumer1除了原有的0~3，会再分配一个7
consumer2除了原有的4~6，会再分配8和9

2.7 producer发布消息机制剖析

1、写入方式
producer 采用 push 模式将消息发布到 broker，每条消息都被 append 到 patition 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。
2、消息路由
producer 发送消息到 broker 时，会根据分区算法选择将其存储到哪一个 partition。其路由机制为：

1. 指定了 patition，则直接使用；
2. 未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition
3. patition 和 key 都未指定，使用轮询选出一个 patition。

3、写入流程

1. producer 先从 zookeeper 的 “/brokers/…/state” 节点找到该 partition 的 leader
2. producer 将消息发送给该 leader
3. leader 将消息写入本地 log
4. followers 从 leader pull 消息，写入本地 log 后 向leader 发送 ACK
5. leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 发送 ACK

2.8 HW与LEO详解

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。

下图详细的说明了当producer生产消息至broker后，ISR以及HW和LEO的流转过程：

由此可见，Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式极大的影响了吞吐率。而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。再回顾下消息发送端对发出消息持久化机制参数acks的设置，我们结合HW和LEO来看下acks=1的情况

结合HW和LEO看下acks=1的情况

2.8日志分段存储

Kafka 一个分区的消息数据对应存储在一个文件夹下，以topic名称+分区号命名，消息在分区内是分段(segment)存储，每个段的消息都存储在不一样的log文件里，这种特性方便old segment file快速被删除，kafka规定了一个段位的 log 文件最大为 1G，做这个限制目的是为了方便把 log 文件加载到内存去操作：

# 部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index文件，
# 如果要定位消息的offset会先在这个文件里快速定位，再去log文件里找具体消息
00000000000000000000.index
# 消息存储文件，主要存offset和消息体
00000000000000000000.log
# 消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset到timeindex文件，
# 如果需要按照时间来定位消息的offset，会先在这个文件里查找
00000000000000000000.timeindex

00000000000005367851.index
00000000000005367851.log
00000000000005367851.timeindex

00000000000009936472.index
00000000000009936472.log
00000000000009936472.timeindex

3 生产问题&&性能调优

3.1 Kafka可视化管理工具kafka-manager

安装教程链接： https://www.cnblogs.com/dadonggg/p/8205302.html

3.2 JVM 参数调优

kafka是scala语言开发，运行在JVM上，需要对JVM参数合理设置，参看JVM调优专题
修改bin/kafka-start-server.sh中的jvm设置，假设机器是32G内存，可以如下设置：

export KAFKA_HEAP_OPTS="‐Xmx16G ‐Xms16G ‐Xmn10G ‐XX:MetaspaceSize=256M ‐XX:+UseG1GC ‐XX:MaxGCPauseMillis=50 ‐XX:G1HeapRegionSize=16M"

# 参数解释
-Xmx最大Heap的大小
-Xms初始的Heap的大小
-Xss每个线程堆栈的大小
-Xmn年轻代大小
‐XX:MetaspaceSize 元空间永久代大小
‐XX:+UseG1GC 垃圾收集器G1
‐XX:MaxGCPauseMillis 每次GC最大的停顿毫秒数
‐XX:G1HeapRegionSize设置的G1 区域的大小

3.3 消息丢失情况

消息发送端：

（1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消
息。大数据统计报表场景，对性能要求很高，对数据丢失不敏感的情况可以用这种。
（2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消
息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
（3）acks=-1或all： 这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志，这种策略会保证只要有一
个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。当然如果
min.insync.replicas配置的是1则也可能丢消息，跟acks=1情况类似。

（4）重试机制 参数设置：

retries = Integer.MAX_VALUE
max.in.flight.requests.per.connection = 1

这样设置后，发送客户端会一直进行重试直到broker返回ack；同时只有一个连接向broker发送数据保证了数据的顺序性。在Leader选举、集群中一个broker挂掉时，发送端会一直重试直到Leader选举结束

消息消费端：
如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了，但是此时你consumer直接宕机了，未处理完的数据丢失了，下次也消费不到了。

解决方案：
(1) enable.auto.commit=false
消费配置设置手动提交
(2) 业务处理失败重试

3.4 消息重复消费

消息发送端：
发送消息如果配置了重试机制，比如网络抖动时间过长导致发送端发送超时，实际broker可能已经接收到消息，但发送方会重新发送消息
消息消费端：
如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理一般消费端都是要做消费幂等处理的。

kafka生产者的幂等性：因为发送端重试导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息只接收一次，只需在生产者加上参数 props.put(“enable.idempotence”, true)即可，默认是false不开启。具体实现原理是，kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker，broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和Sequence Number，如果相同不会再接收。

3.5 消息乱序

如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现，发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1了所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息，当然acks不能设置为0，这样也能保证消息从发送
端到消费端全链路有序。kafka保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。

3.6 消息积压

1）线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。
此种情况如果积压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到的消息快速转发到其他topic(可以设置很多分区)，然后再启动多个消费者同时消费新主题的不同分区。
2）由于消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。
此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)，后面再慢慢分析死信队列里的消息处理问题。