浅谈Kafka（二）

Kafka架构图

1. 生产消费模型涉及到生产者、消费者和消息队列。Kafka消息度列有点对点模式和发布订阅模式。

• 点对点模式，生产者生产消息发送到消息队列中，然后消费者组从消息队列中取出并消费消息。消息被消费后，消息队列中不再有存储，所以消费者不可能消费到已经被消费的消息。特点是每个消息只有一个消费者，一旦被消费，消息就不再存储在消息队列中；生产者和消费者之间没有依赖性，生产者发送消息后，不管消费者在运行，都不会影响生产者再次发送消息；消费者成功收到消息后需要向消息队列应答成功，以便消息队列删除当前接收到的消息。
• 发布订阅模式，每个消息可以有多个订阅者；发布者和订阅者之间有时间上的限制。对于某个主题的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息。为了消费消息，订阅者需要提前订阅该主题，并保持在线运行。

2. Kafka可以构建实时数据管道，以可靠地在系统或应用程序之间获取数据；还可以用来构建实时流应用程序，以转换或响应数据流。

3. 一个Kafka集群是由多个broker组成，这样可以实现负载均衡以及容错；broker是无状态的，它们之间通过zookeeper来维护集群状态；一个broker每秒可以处理数十万次读写，每个broker都可以处理TB消息而不影响性能。

4. zookeeper可以管理和协调broker并且存储了Kafka的元数据，主要用于通知生产者和消费者Kafka集群有新的broker加入或者有broker出现了故障。

5. 生产者producer负责把数据推送到broker的topic，消费者consumer负责从broker的topic中拉取消息并自行处理，Kafka集群中主题被分为多个分区partition。副本replication可以确保某个服务故障时数据依然可用，一般副本数大于1。消费者组是可扩展且具有容错的消费者机制，可以包含多个消费者，具有唯一的ID，组内的消费者一起消费主题的所有分区。

6. 主题是一个逻辑概念，用于生产者发布消息，消费者拉取消息。Kafka中的主题必须要有标识符，并且是唯一的，可以有任意数量的主题。在主题中的消息是有结构的，一般一个主题包含某一类消息。一旦生产者发送消息到主题中，这些消息就不能更改。

7. 偏移量offset记录着下一条要发送给消费者的消息的序号，默认Kafka把offset存储在zookeeper中。一个分区中消息的存储是有顺序的，每个分区的消息都是有一个递增的id就是offset。偏移量在分区里才是有意义的。

Kafka生产者幂等性与事务

1. 幂等性比如http的一次或多次请求得到的响应是一样的。在生产者生产消息时，如果出现重试时，可能一条消息被发送了多次，如果不具备幂等性，就有可能在分区中保存多条一模一样的消息。配置幂等性，设置enable.idempotence选项为true。

2. 幂等性原理，通过引入Producer ID(PID)和Sequence Number。每个生产者在初始化时，都会分配一个唯一的PID，这个PID对用户来说是透明的。针对每个生产者发送到指定主题分区的消息都对应一个从0开始递增的序列号。

3. 生产者消息重复问题，在Kafka中可以开启幂等性，当生产者生产消息时，会增加一个PID和序列号。发送消息时，会连着PID和序列号一起发出。Kafka收到消息后，会把消息、PID和序列号一起保存下来。如果ACK响应失败，生产者重试，再次发送消息时，Kafka会根据PID和序列号确定是否需要再保存一条消息，判断条件是生产者发送过来的序列号是否小于等于分区中消息对应的序列号。

生产者分区写入策略

1. 轮询策略是默认的策略，也是使用最多的策略，可以最大限度保证所有消息平均分配到某一个分区。如果在生产消息时键为空，则使用轮询算法均衡地分配分区。
2. 随机分区每次都随机将消息分配到每个分区，基本不用。
3. 按key分区有可能出现数据倾斜，key的哈希值对分区数量取余。
4. 自定义分区实现Partitioner接口，创建自定义分区器，生产者配置自定义的分区器。

乱序问题

1. 轮询和随机策略都会导致一个问题，生产到Kafka中的数据是乱序存储的，而按key分区可以一定程度上实现数据的有序存储，也就是局部有序，但是可能会导致数据倾斜，在实际生产环境中要结合实际情况做取舍。
2. 生产者是由写入策略，如果主题有多个分区，就会把数据分散在不同的分区中存储。
3. 当分区数量大于1时，消息会打散分不到不同的分区中。如果只有一个分区，消息是有序的。

消费者组的Reblance机制

1. Kafka再平衡是确保消费者下所有消费者如何达成一致分配订阅的主题的每个分区的机制。
2. 再平衡的触发时机包括消费者组中消费者的个数发生变化，订阅的主题个数发生变化，订阅的主题分区数发生变化。
3. 发生再平衡时，消费者组下所有的消费者都会协调在一起共同参与，Kafka使用分配策略尽可能达到最公平的分配。再平衡过程会对消费者组产生非常严重的影响，所有的消费者都将停止工作，直到再平衡完成。

消费者分区分配策略

1. Range范围分配策略是默认分配策略，可以确保每个消费者消费的分区数量是均衡的。范围分配策略是针对每个topic的。配置消费者的partition.assignment.strategy为RangeAssignor。

算法公式：n = 分区数量/消费者数量；
m = 分区数量 % 消费者数量；
前m个消费者消费n+1个，剩余消费者消费n个。

2. RoundRobin轮询分配策略按照字典序排序，然后通过轮询方式逐个把分区依次分配给每个消费者。配置分配策略为RoundRobinAssignor。

3. Sticky粘性分配策略分区分配尽可能均匀，发生再平衡时，分区的分配尽可能与上一次分配保持相同。没有发生再平衡时，粘性分配策略和轮询策略类似。

副本机制

1. 副本的目的就是冗余备份。当某个broker上的分区数据丢失时，依然可以保证数据可用。因为在其他的broker上的副本是可用的。
2. 对副本关系较大的就是producer配置的acks参数了，acks参数表示当生产者生产消息的时候，写入到副本的要求严格程度。它决定了如何在性能和可靠性之间做取舍。
3. acks配置为0：不等待broker确认，直接发送下一条数据。性能最高，但可能会存在数据丢失的情况。
   acks配置为1：等待leader副本确认接收后，才会发送下一条数据，性能中等。
   acks配置为-1或all：等待所有副本已经将数据同步后，才会发送下一条数据，性能最差。

分区的leader与follower

1. Kafka中leader负责处理读写操作，而follower只负责副本数据的同步。
2. 如果leader出现故障，其他follower会被重新选举为leader。
3. follower向一个消费者一样，拉取leader对应分区的数据并保存到日志中。

AR/ISR/OSR

1. 分区的所有副本称为AR已分配的分区。
2. 所有与leader副本保持一定程度同步的副本组成ISR(In Sync Replica在同步中的副本)。
3. 由于follower副本同步之后过多的副本(不包括leader副本)组成OSR。
4. 正常情况下，所有的follower副本都应该与leader副本保持同步。

controller介绍与选举

1. 在Kafka启动时会在所有的broker中选择一个controller，leader与follower是针对partition，而controller是针对broker的。创建主题、添加分区、修改副本数量之类的管理任务都是由controller来完成的。Kafka分区leader的选举也是由controller决定的。
2. controller选举流程，在集群启动时每个broker都会尝试去zookeeper上注册成为controller(临时节点)。组只有一个竞争成功，其他的broker会注册该节点的监视器。一旦该节点状态发生改变，就可以进行相应的处理。controller也是高可用的，一旦某个broker崩溃，其他broker会重新注册成controller。
3. controller是高可用的，通过zookeeper进行选举；Leader是通过ISR进行快速选举。

Leader负载均衡

1. Kafka中引入了优先副本的概念，在ISR列表中第一个副本就是优先副本。第一个分区存放的broker，肯定就是优先副本。通过脚本实现均匀分配每个分区的leader。

./kafka-leader-election.sh --bootstrap.server localhsot:9092 --topic name --partition=1 --election-type prefered

Kafka读写流程

1. 写过程：通过zookeeper找分区对应的Leader，由Leader负责写数据。Producer开始生产数据，ISR里面的Follower开始同步数据，并返回给Leader ACK。Leader接收到所有的ISR中的副本的ACK后返回给Producer ACK。

2. 读流程：通过zookeeper找分区对应Leader进行读操作，找到分区对应的偏移量，然后从偏移量往后顺序拉取数据，提交偏移量。

消费者读数据流程

1. 消费者的offset是一个分区的全局offset，存储在zookeeper中，可以通过offset找到对应的segment，然后把全局的offset转换成segment的局部offset。根据局部offset，可以从稀疏索引中找到对应的数据位置，开始顺序读取。

消息不丢失机制

1. broker数据不丢失

• 生产者通过分区的leader写入数据后，所有在ISR中的follower都会从leader中复制数据，这样可以确保即使leader崩溃了，其他的follower的数据依然是可用的。

2. 生产者数据不丢失

• 生产者连接leader写入数据时，通过ACK机制来确保数据写入成功。ACK机制有三个可选配置：

  acks为-1，表示所有的节点都接收到数据（leader和follower都接收到数据）；
  acks为1，表示leader收到数据；
  acks为0，生产者只负责发送数据，不关心数据是否丢失。

• 生产者可以采用同步和异步方式发送数据。同步是发送数据后，等待返回结果。异步是发送数据后，只提供一个回调函数。

• 如果broker迟迟不给ack，而buffer又满了，可以设置是否清空buffer中的数据。

3. 消费者数据不丢失

• 在消费者消费数据时，只要每个消费者记录好offset值即可，就能保证数据不丢失。
• 可以使用MySQL事务，将写入的MySQL数据和offset放在一个MySQL事务里，要么全部成功，要么全部失败，就可以实现Exactly-Once。

数据积压

1. Kafka消费者消费数据的速度是非常快的，但如果处理Kafka消息时，由于有一些外部IO、或者是产生网络拥堵，就会造成Kafka中的数据积压。如果数据一直积压，会较大地影响数据的实时性。
2. 首先找到数据积压的原因。常见的场景包括数据写入MySQL失败，网络延迟消费失败。

Kafka中数据清理

1. Kafka提供两种日志清理方式

• 日志删除(Log Deletion)按照指定的策略直接删除不符合条件的日志。
• 日志压缩(Log Compaction)按照消息的key进行整合，有相同key的但有不同value值，只保留最后一个版本。

2. 在Kafka的broker或topic配置

配置项	配置值	说明
log.cleaner.enable	true(默认)	开启自动清理日志功能
log.cleanup.policy	delete(默认)	删除日志
log.cleanup.policy	compaction	压缩日志
log.cleanup.policy	delete.compact	同时支持删除、压缩

3. 日志删除是以段为单位进行定期清理的。Kafka日志管理器有一个专门的日志删除任务定期检测和删除不符合保留条件的日志分段文件，这个周期可以通过broker的参数log.retention.check.interval.ms来配置，默认5分钟。日志分段的保留策略有基于时间的保留策略、基于日志大小的保留策略和基于日志起始偏移量的保留策略。

4. 基于时间的保留策略有log.retention.ms、log.retention.minutes、log.retentions.hours三种配置，优先级由高到低，默认168小时即7天。删除日志分段时，从日志文件对象中所维护日志分段的跳跃表中移除待删除日志分段，以保证没有线程对这些日志分段进行读取操作；将日志分段文件添加上.deleted的后缀；Kafka后台定时任务会定期删除这些.deleted后缀的文件，这个任务的延迟执行时间可以通过file.delete.delay.ms参数来设置，默认1分钟。