Apache Kafka学习

一、简介

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。Kafka是一种消息队列，主要用来处理大量数据状态下的消息队列，一般用来做日志的处理。

官方中文文档 Kafka 中文文档 - ApacheCN

1.概念：

1.Kafka作为一个集群，运行在一台或者多台服务器上

2.Kafka 通过 topic 对存储的流数据进行分类

3.每条记录中包含一个key，一个value和一个timestamp（时间戳）

2.kafka四大API：

Producer API，它允许应用程序向一个或多个 topics 上发送消息记录

Consumer API，允许应用程序订阅一个或多个 topics 并处理为其生成的记录流

Streams API，它允许应用程序作为流处理器，从一个或多个主题中消费输入流并为其生成输出流，有效的将输入流转换为输出流。

Connector API，它允许构建和运行将 Kafka 主题连接到现有应用程序或数据系统的可用生产者和消费者。例如，关系数据库的连接器可能会捕获对表的所有更改

3.Kafka消费模式

一对一的消费，也即点对点的通信，即一个发送一个接收。消息生产者发布消息到Queue队列中，通知消费者从队列中拉取消息进行消费。消息被消费之后则删除，Queue支持多个消费者，但对于一条消息而言，只有一个消费者可以消费，即一条消息只能被一个消费者消费。

一对多的消费，即一个消息发送到消息队列，消费者根据消息队列的订阅拉取消息消费。发布/订阅模式，即利用Topic存储消息，消息生产者将消息发布到Topic中，同时有多个消费者订阅此topic，订阅了这个topic的消费者都能消费这条消息，注意发布到Topic中的消息会被多个消费者消费，消费者消费数据之后，数据不会被清除，Kafka会默认保留一段时间，然后再删除。 

4.Kafka的基础架构

如上图所示，一个典型的Kafka集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

5.kafka文件存储方式

kafka存储的数据是以追加的方式添加到队列尾部。读写数据是顺序读写。

由于生产者生产的消息会不断追加到 log 文件末尾， 为防止 log 文件过大导致数据定位效率低下， Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。

每个 segment对应两个文件——“.index”文件和“.log”文件。 这些文件位于一个文件夹下， 该文件夹的命名规则为： topic 名称+分区序号。例如， first 这个 topic 有三个分区，则其对应的文件夹为 first-0,first-1,first-2。

二、特性

流处理平台有以下三种特性:

1. 可以让你发布和订阅流式的记录。这一方面与消息队列或者企业消息系统类似。
2. 可以储存流式的记录，并且有较好的容错性。
3. 可以在流式记录产生时就进行处理。

kafka特性：

1. 通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能
2. 高吞吐量 [2]  ：即使是非常普通的硬件Kafka也可以支持每秒数百万 [2]  的消息
3. 支持通过Kafka服务器和消费机集群来分区消息
4. 支持Hadoop并行数据加载

三、优点

1.解耦

耦合的状态表示当你实现某个功能的时候，是直接接入当前接口。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

正常链路：A调用->B

A发送->MQ订阅->B

2.异步处理

异步处理替代了之前的同步处理，异步处理不需要让流程走完就返回结果，可以将消息发送到消息队列中，然后返回结果，剩下让其他业务处理接口从消息队列中拉取消费处理即可。

3.流量削峰

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。高流量的时候，使用消息队列作为中间件可以将流量的高峰保存在消息队列中，从而防止了系统的高请求，减轻服务器的请求处理压力。

4.数据持久化

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

5.顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

6.可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

四、名词解释

1.Producer ： 消息生产者，就是向 Kafka发送数据 

2.Consumer ： 消息消费者，从 Kafka broker中读取数据

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

4.Broker ：经纪人，一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多个 topic

5.Topic ： 主题，可以理解为一个队列， 生产者和消费者面向的都是一个 topic

6.Partition：分区，为了实现扩展性，一个非常大的Topic可以分布到多个Broker上，一个Topic可以分为多个Partition，每个Partition是一个有序的队列(分区有序，不能保证全局有序)

7.Replica： 副本（Replication），为保证集群中的某个节点发生故障时， 该节点上的 partition 数据不丢失，且 Kafka仍然能够继续工作， Kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower

8.Leader： 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 leader

9.Follower： 每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。 leader 发生故障时，某个 Follower 会成为新的 leader

10.Offset：每个Consumer 消费的信息都会有自己的序号，我们称作当前队列的offset。即消费点位标识消费到的位置。每个消费组都会维护订阅的Topic 下每个队列的offset

五、QA

Q:如何保证数据高可靠、不丢失?

A：数据丢失的原因

1. 生产消息阶段  如果出现了网络不可用、消息本身不合格等原因导致消息根本没有被 Broker 接收，那就相当于消息在生产者端就消失了。
2. 存储消息阶段 Broker 端的消息丢失，一般是由 Broker 服务不可用造成的，例如 Broker 都宕机了导致消息丢失
3. 消费消息阶段 消费者在消费消息的过程中，会同时更新消费者位移，也就是「已经消费到哪一条消息了」。这里就存在一个问题，当消费一个消息的时候，是先处理消息，成功后再更新位移，还是先更新位移，再处理消息。 如果先更新位移，在处理消息，当消息处理出现问题，或者更新完位移、消息还未处理，消费者出现宕机等问题的时候，消息就会丢失。 而如果先处理消息再更新位移，虽然可能会出现重复消费同一个消息的问题，但是，我们可以通过消费者处理逻辑实现幂等的方式来解决。

解决方案

producer 生产消息

ack 机制

生产者 acks参数指定了必须要有多少个分区副本收到消息，生产者才认为该消息是写入成功的，这个参数对于消息是否丢失起着重要作用。

ack 策略

现在我们已经知道生产者发送消息有个确认的机制，那么Kafka里是何时确认呢？Kafka是通过配置acks的值确认机制的，这里一共提供了三种策略，对应不同的ACK机制：

1. acks=0，生产者不等待broker的响应。这种情况下延迟最低，但是有可能丢失数据，比较适合高吞吐量、接受消息丢失的场景。
2. acks=1，生产者发送消息等待broker的响应，等待leader落盘成功后响应确认。这种情况下，如果是在leader完成同步消息给follower前发生故障，则可能发生消息丢失。
3. acks=-1，生产者发送消息等待broker的响应，直到leader和follower全部落盘成功后才会响应确认。此机制能严格保证不丢失数据。但当所有的follower同步完成之后，leader发送ack响应之前，leader发生了宕机，此时生产者会以为发送失败了，然后会重新发送数据给新的leader，因此该情况下会导致数据重复发送。

broker 存储消息

存储消息阶段需要在消息刷盘之后再给生产者响应，假设消息写入缓存中就返回响应，那么机器突然断电这消息就没了，而生产者以为已经发送成功了。

如果Broker是集群部署，有多副本机制，即消息不仅仅要写入当前,还需要写入副本机中。那配置成至少写入两台机子后再给生产者响应。这样基本上就能保证存储的可靠了。所以broker 消息存储主要是靠的是冗余副本，即多个Replica

ISR机制 和 AR机制

简单来说，分区中的所有副本统称为 AR (Assigned Replicas)。所有与leader副本保持一定程度同步的副本（包括leader副本在内）组成 ISR (In Sync Replicas)。 ISR 集合是 AR 集合的一个子集。消息会先发送到leader副本，然后follower副本才能从leader中拉取消息进行同步。同步期间，follow副本相对于leader副本而言会有一定程度的滞后。前面所说的 ”一定程度同步“ 是指可忍受的滞后范围，这个范围可以通过参数进行配置。于leader副本同步滞后过多的副本（不包括leader副本）将组成 OSR （Out-of-Sync Replied）由此可见，AR = ISR + OSR。正常情况下，所有的follower副本都应该与leader 副本保持 一定程度的同步，即AR=ISR，OSR集合为空。

leader副本负责维护和跟踪 集合中所有follower副本的滞后状态，当follower副本落后太多或失效时，即follower长时间未向leader发送消息，leader副本会把它从 ISR 集合中剔除。如果 集合中所有follower副本“追上”了leader副本，那么leader副本会把它从 OSR 集合转移至 ISR 集合【副本可以在OSP,ISR中来回移动】。默认情况下，当leader副本发生故障时，只有在 ISR 集合中的follower副本才有资格被选举为新的leader，而在 OSR 集合中的副本则没有任何机会（不过这个可以通过配置来改变）。

broker恢复机制

• LEO：（Log End Offset）每个副本的最后一个offset
• HW：（High Watermark）高水位，指的是消费者能见到的最大的 offset， ISR 队列中最小的 LEO。可以理解为短板效应

follower 故障：follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后， follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

leader 故障：leader 发生故障之后，会从 ISR 中选出一个新的 leader，之后，为保证多个副本之间的数据一致性， 其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader同步数据。

comsumer 消费消息

消费者拿到消息之后直接存入内存队列中就直接返回给消费成功，这样其实是不算消息消费成功的。我们需要考虑消息放在内存之后消费者就宕机了怎么办，若直接设置为消费成功，当前情况下本条消息相当于丢失了。

所以我们应该在消费者真正执行完业务逻辑之后，再发送给消费成功，这才是真正的消费了。