[大数据]kafka总结

Kafka 简介

Apache Kafka起源于LinkedIn，后来于2011年成为开源Apache项目，然后于2012年成为First-class Apache项目。Kafka是用Scala和Java编写的。 Apache Kafka是基于发布订阅的容错消息系统。 它是快速，可扩展和设计分布。

Kafka特点：

• 可靠性：具有副本及容错机制。
• 可扩展性：kafka无需停机即可扩展节点及节点上线。
• 持久性：数据存储到磁盘上，持久性保存。
• 性能：kafka具有高吞吐量。达到TB级的数据，也有非常稳定的性能。
• 速度快：顺序写入和零拷贝技术使得kafka延迟控制在毫秒级。

kafka架构

kafka支持消息持久化，消费端是主动拉取数据，消费状态和订阅关系由客户端负责维护，消息消费完后，不会立即删除，会保留历史消息。因此支持多订阅时，消息只会存储一份就可以。

broker

• kafka集群中包含一个或者多个服务实例（节点），这种服务实例被称为broker（一个broker就是一个节点/一个服务器）

topic

• topic特指kafka处理的消息源（feeds of messages）的不同分类，每条发布到kafka集群的消息都属于某个类别
• topic是一种分类或者发布的一些列记录的名义上的名字。kafka主题始终是支持多用户订阅的；也就是说，一 个主题可以有零个，一个或者多个消费者订阅写入的数据
• 在kafka集群中，可以有无数的主题
• 生产者和消费者消费数据一般以主题为单位。更细粒度可以到分区级别。

partition

• partition是一个物理上的概念，每个topic包含一个或者多个partition
• kafka当中，topic是消息的归类，一个topic可以有多个分区（partition），每个分区保存部分topic的数据，所有的partition当中的数据全部合并起来，就是一个topic当中的所有的数据。
• 一个broker服务下，可以创建多个分区，broker数与分区数没有关系
• 在kafka中，每一个分区会有一个编号：编号从0开始
• 每一个分区内的数据是有序的，但全局的数据不能保证是有序的（有序是指生产什么样顺序，消费时也是什么样的顺序）

segment

• 一个partition当中存在多个segment文件段，每个segment分为两部分，.log文件和 .index 文件

  • .log：存放数据文件，文件当中数据的偏移量位置；
  • .index：存放.log文件的索引数据，主要用于快速查询。

• 索引文件与数据文件的关系

  • 索引文件中元数据指向对应数据文件中message的物理偏移地址。
  • 索引文件，里面存储的是一对一对的key-value，其中key是消息在对应log文件第几条数据；value 代表的是该消息物理偏移地址以及在全局partiton中的第几个消息（顺序写入特性）。
  • 比如索引文件中 3,497 代表：数据文件中的第三个message，它的偏移地址为497，也表示在全局partiton表示第497个消息。数据文件中，Message 368772表示：在全局partiton中是第368772个message。

  那么为什么在index文件中这些编号不是连续的呢？
  这是因为index文件中并没有为数据文件中的每条消息都建立索引，而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引，减少索引文件大小，从而可以将索引文件保留在内存中，通过mmap（内存映射）可以直接内存操作。
  稀疏索引比稠密索引减少大小占用，但查找速度下降，没有建立索引的Message也不能一次定位到其在数据文件的位置，从而需要做一次顺序扫描，但是这次顺序扫描的范围就很小了。

  

• log、index文件目录及组成

  • kafka在我们指定的log.dir目录下，会创建一些文件夹；名字是 （主题名字-分区名） 所组成的文件夹。 在（主题名字-分区名）的目录下，会有两个文件存在，如下所示：

    #索引文件
    00000000000000000000.index
    #日志内容
    00000000000000000000.log
    

  • 在目录下的文件，会根据log日志的大小进行切分，.log文件的大小为1G的时候，就会进行切分文件；如下：

    -rw-r--r--. 1 root root 389k  1月  17  18:03   00000000000000000000.index
    -rw-r--r--. 1 root root 1.0G  1月  17  18:03   00000000000000000000.log
    -rw-r--r--. 1 root root  10M  1月  17  18:03   00000000000000077894.index
    -rw-r--r--. 1 root root 127M  1月  17  18:03   00000000000000077894.log
    

  • segment文件命名规则：partion全局的第一个segment从0开始，后续每个segment文件名为上一个全局 partion的最大offset（偏移message数）。数值最大为64位long大小，20位数字字符长度，没有数字就用 0 填充。

producer

• producer主要是用于生产消息，是kafka当中的消息生产者，生产的消息通过topic进行归类，保存到kafka的broker里面去。

consumer

• 消息的消费者，向 kafka 的 broker 中读取消息的客户端；消费者一定是归属于某个消费组中的。

consumer group

• 消费者组，每一个 consumer 属于一个特定的 consumer group（可以为每个consumer指定 groupName）
• 消费者组由一个或者多个消费者组成，同一个组中的消费者对于同一条消息只消费一次。
• 每个消费者都属于某个消费者组，如果不指定，那么所有的消费者都属于默认的组。
• 每个消费者组都有一个ID，即group ID。组内的所有消费者协调在一起来消费一个订阅主题( topic)的所有分区(partition)。当然，每个分区只能由同一个消费组内的一个消费者(consumer)来消费，可以由不同的消费组来消费。
• partition数量决定了每个consumer group中并发消费者的最大数量。如下图：如果只有两个分区，即使一个组内的消费者有4个，也会有两个空闲的。有4个分区，每个消费者消费一个分区，并发量达到最大4。
  
• 总结：
  • 某一个主题下的分区数，对于消费该主题的同一个消费组下的消费者数量，应该小于等于该主题下的分区数。最好与分区数成整数倍。同一个分区下的数据，在同一时刻，不能同一个消费组的不同消费者消费。
  • 分区数越多，同一时间可以有越多的消费者来进行消费，消费数据的速度就会越快，提高消费的性能。

partition replicas（分区副本）

• kafka 中的分区副本如下图所示：
  

• 副本数、副本因子（replication-factor）：控制消息保存在几个broker（服务器）上，一般情况下副本数等于broker的个数。

• 一个broker服务下，不可以创建多个副本因子。创建主题时，副本因子应该小于等于可用的broker数。

• 副本因子操作以分区为单位的。每个分区都有各自的主副本和从副本；

• 主副本叫做leader，从副本叫做 follower（在有多个副本的情况下，kafka会为同一个分区下的所有分区，设定角色关系：一个leader和N个 follower），处于同步状态的副本叫做in-sync-replicas(ISR)，这也是表示副本可用

• follower通过拉的方式从leader同步数据。

• 消费者和生产者都是从leader读写数据，不与follower交互。

• 副本因子的作用：让kafka读取数据和写入数据时的可靠性。

• 副本因子是包含本身，同一个副本因子不能放在同一个broker中。

• 如果某一个分区有三个副本因子，就算其中一个挂掉，那么只会剩下的两个中，选择一个leader，但不会在其他的broker中，另启动一个副本（因为在另一台启动的话，存在数据传递，只要在机器之间有数据传递，就会长时间占用网络IO，kafka是一个高吞吐量的消息系统，这个情况不允许发生）所以不会在另一个broker中启动。

• 如果所有的副本都挂了，生产者如果生产数据到指定分区的话，将写入不成功。

为什么要使用 kafka？

缓冲和削峰：上游数据时有突发流量，下游可能扛不住，或者下游没有足够多的机器来保证冗余，kafka在中间可以起到一个缓冲的作用，把消息暂存在kafka中，下游服务就可以按照自己的节奏进行慢慢处理。

解耦和扩展性：项目开始的时候，并不能确定具体需求。消息队列可以作为一个接口层，解耦重要的业务流程。只需要遵守约定，针对数据编程即可获取扩展能力。

冗余：可以采用一对多的方式，一个生产者发布消息，可以被多个订阅topic的服务消费到，供多个毫无关联的业务使用。

健壮性：消息队列可以堆积请求，所以消费端业务即使短时间死掉，也不会影响主要业务的正常进行。

异步通信：很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

Kafka消费过的消息如何再消费？

kafka消费消息的offset是定义在zookeeper中的， 如果想重复消费kafka的消息，可以在redis中自己记录offset的checkpoint点（n个），当想重复消费消息时，通过读取redis中的checkpoint点进行zookeeper的offset重设，这样就可以达到重复消费消息的目的了

kafka的数据是放在磁盘上还是内存上，为什么速度会快？

kafka使用的是磁盘存储。

速度快是因为：

1. 顺序写入：因为硬盘是机械结构，每次读写都会寻址->写入，其中寻址是一个“机械动作”，它是耗时的。所以硬盘 “讨厌”随机I/O， 喜欢顺序I/O。为了提高读写硬盘的速度，Kafka就是使用顺序I/O。
2. Memory Mapped Files（内存映射文件）：64位操作系统中一般可以表示20G的数据文件，它的工作原理是直接利用操作系统的Page来实现文件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上。
3. Kafka高效文件存储设计： Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。通过索引信息可以快速定位
   message和确定response的大小。通过index元数据全部映射到memory（内存映射文件），可以避免segment file的IO磁盘操作。通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小。

注：1. Kafka解决查询效率的手段之一是将数据文件分段，比如有100条Message，它们的offset是从0到99。假设将数据文件分成5段，第一段为0-19，第二段为20-39，以此类推，每段放在一个单独的数据文件里面，数据文件以该段中的offset命名。这样在查找指定offset的Message的时候，用二分查找就可以定位到该Message在哪个段中。
2.为数据文件建索引数据文件分段使得可以在一个较小的数据文件中查找对应offset的Message了，但是这依然需要顺序扫描才能找到对应offset的Message。为了进一步提高查找的效率，Kafka为每个分段后的数据文件建立了索引文件，文件名与数据文件的名字是一样的，只是文件扩展名为.index。

Kafka数据怎么保障不丢失？

分三个点说，一个是生产者端，一个消费者端，一个broker端。

1.生产者数据的不丢失

kafka的ack机制

• 在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的能够被收到，其中状态有0，1，-1。
• ack=0：producer不等待broker同步完成的确认，继续发送下一条(批)信息。
• ack=1（默认）：producer要等待leader成功收到数据并得到确认，才发送下一条message。
• ack=-1：producer得到follwer确认，才发送下一条数据。

生产者发送数据有同步方式和异步方式

同步方式

• 发送一批数据给kafka后，等待kafka返回结果：
  • 生产者等待10s，如果broker没有给出ack响应，就认为失败。
  • 生产者重试3次，如果还没有响应，就报错.
• 同步方式下，ack设置为0，风险很大，一般不建议设置为0。即使设置为1，也会随着leader宕机丢失数据。所以如果要严格保证生产端数据不丢失，可设置为-1。

异步方式

• 发送一批数据给kafka，只是提供一个回调函数：
  • 先将数据保存在生产者端的buffer中。buffer大小是2万条 。
  • 满足数据阈值或者数量阈值其中的一个条件就可以发送数据。
  • 发送一批数据的大小是500条。
  • 如果broker迟迟不给ack，而buffer又满了，开发者可以设置是否直接清空buffer中的数据。
• 异步方式下，也会考虑ack的状态，除此之外，异步模式下的有个buffer，通过buffer来进行控制数据的发送，有两个值来进行控制，时间阈值与消息的数量阈值，如果buffer满了数据还没有发送出去，有个选项是配置是否立即清空buffer。可以设置为-1，永久阻塞，也就数据不再生产。异步模式下，即使设置为-1。也可能因为程序员的不科学操作，操作数据丢失，比如kill -9，但这是特别的例外情况。

2.消费者数据的不丢失

通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费。

而offset的信息在kafka0.8版本之前保存在zookeeper中，在0.8版本之后保存到topic中，即使消费者在运行过程中挂掉了，再次启动的时候会找到offset的值，找到之前消费消息的位置，接着消费，由于 offset 的信息写入的时候并不是每条消息消费完成后都写入的，所以这种情况有可能会造成重复消费，但是不会丢失消息。

唯一例外的情况是，我们在程序中给原本做不同功能的两个consumer组设置
KafkaSpoutConfig.bulider.setGroupid的时候设置成了一样的groupid，这种情况会导致这两个组共享同一份数据，就会产生组A消费partition1，partition2中的消息，组B消费partition3的消息，这样每个组消费的消息都会丢失，都是不完整的。 为了保证每个组都独享一份消息数据，groupid一定不要重复才行。

3.broker的数据不丢失

每个broker中的partition我们一般都会设置有replication（副本）的个数，生产者写入的时候首先根据分发策略（有partition按partition，有key按key，都没有轮询）写入到leader中，follower（副本）再跟leader同步数据，这样有了备份，也可以保证消息数据的不丢失。

采集数据为什么选择kafka？

采集层 主要可以使用Flume, Kafka等技术。

Flume：Flume 是管道流方式，提供了很多的默认实现，让用户通过参数部署，及扩展API.

Kafka：Kafka是一个可持久化的分布式的消息队列。 Kafka 是一个非常通用的系统。你可以有许多生产者和很多的消费者共享多个主题Topics。

相比之下,Flume是一个专用工具被设计为旨在往HDFS，HBase发送数据。它对HDFS有特殊的优化，并且集成了Hadoop的安全特性。

所以，Cloudera 建议如果数据被多个系统消费的话，使用kafka；如果数据被设计给Hadoop使用，使用Flume。

kafka 重启是否会导致数据丢失？

kafka是将数据写到磁盘的，一般数据不会丢失。

但是在重启kafka过程中，如果有消费者消费消息，那么kafka如果来不及提交offset，可能会造成数据的不准确（丢失或者重复消费）。

kafka 宕机了如何解决？

先考虑业务是否受到影响
kafka 宕机了，首先我们考虑的问题应该是所提供的服务是否因为宕机的机器而受到影响，如果服务提供没问题，如果实现做好了集群的容灾机制，那么这块就不用担心了。

节点排错与恢复
想要恢复集群的节点，主要的步骤就是通过日志分析来查看节点宕机的原因，从而解决，重新恢复节点。

为什么Kafka不支持读写分离？

在 Kafka 中，生产者写入消息、消费者读取消息的操作都是与 leader 副本进行交互的，从 而实现的是一种主写主读的生产消费模型。

Kafka 并不支持主写从读，因为主写从读有 2 个很明显的缺点:

• 数据一致性问题：数据从主节点转到从节点必然会有一个延时的时间窗口，这个时间 窗口会导致主从节点之间的数据不一致。某一时刻，在主节点和从节点中 A 数据的值都为 X， 之后将主节点中 A 的值修改为 Y，那么在这个变更通知到从节点之前，应用读取从节点中的 A 数据的值并不为最新的 Y，由此便产生了数据不一致的问题。

• 延时问题：类似 Redis 这种组件，数据从写入主节点到同步至从节点中的过程需要经历 网络→主节点内存→网络→从节点内存 这几个阶段，整个过程会耗费一定的时间。而在 Kafka 中，主从同步会比 Redis 更加耗时，它需要经历 网络→主节点内存→主节点磁盘→网络→从节 点内存→从节点磁盘 这几个阶段。对延时敏感的应用而言，主写从读的功能并不太适用。

而kafka的主写主读的优点就很多了：

• 可以简化代码的实现逻辑，减少出错的可能;
• 将负载粒度细化均摊，与主写从读相比，不仅负载效能更好，而且对用户可控;
• 没有延时的影响;
• 在副本稳定的情况下，不会出现数据不一致的情况。

kafka数据分区和消费者的关系？

每个分区只能由同一个消费组内的一个消费者(consumer)来消费，可以由不同的消费组的消费者来消费，同组的消费者则起到并发的效果。

kafka的数据offset读取流程

1. 连接ZK集群，从ZK中拿到对应topic的partition信息和partition的Leader的相关信息
2. 连接到对应Leader对应的broker
3. consumer将?自?己保存的offset发送给Leader
4. Leader根据offset等信息定位到segment（索引?文件和?日志?文件）
5. 根据索引?文件中的内容，定位到?日志?文件中该偏移量量对应的开始位置读取相应?长度的数据并返回给consumer

kafka内部如何保证顺序，结合外部组件如何保证消费者的顺序？

kafka只能保证partition内是有序的，但是partition间的有序是没办法的。爱奇艺的搜索架构，是从业务上把需要有序的打到同?个partition。

Kafka消息数据积压，Kafka消费能力不足怎么处理？

1. 如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数=分区数。（两者缺一不可）

2. 如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间<生产速度），使处理的数据小于生产的数据，也会造成数据积压。

Kafka单条日志传输大小

kafka对于消息体的大小默认为单条最大值是1M但是在我们应用场景中, 常常会出现一条消息大于1M，如果不对kafka进行配置。则会出现生产者无法将消息推送到kafka或消费者无法去消费kafka里面的数据, 这时我们就要对kafka进行以下配置：server.properties

replica.fetch.max.bytes: 1048576  #broker可复制的消息的最大字节数, 默认为1M
message.max.bytes: 1000012   #kafka 会接收单个消息size的最大限制， 默认为1M左右

注意：message.max.bytes必须小于等于replica.fetch.max.bytes，否则就会导致replica之间数据同步失败。