Kafka的相关知识-创新互联

一. Kafka基本介绍

Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统。具有：高吞吐量、低延迟、可扩展性、持久性、可靠性、容错性、高并发等特性。常见的应用场景有：日志收集、消息系统、流式处理等。

创新互联坚持“要么做到，要么别承诺”的工作理念，服务领域包括：成都网站建设、网站建设、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的灵川网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！二. Kafka的基本架构

• Producer：生产者，也就是发送消息的一方。生产者负责创建消息，然后将其发送到 Kafka。
• Consumer：消费者，也就是接受消息的一方。消费者连接到 Kafka 上并接收消息，进而进行相应的业务逻辑处理。
• Consumer Group：一个消费者组可以包含一个或多个消费者。使用多分区 + 多消费者方式可以极大提高数据下游的处理速度，同一消费组中的消费者不会重复消费消息，同样的，不同消费组中的消费者消息消息时互不影响。Kafka 就是通过消费组的方式来实现消息 P2P 模式和广播模式。
• Broker：服务代理节点。Broker 是 Kafka 的服务节点，即 Kafka 的服务器。
• Topic：Kafka 中的消息以 Topic 为单位进行划分，生产者将消息发送到特定的 Topic，而消费者负责订阅 Topic 的消息并进行消费。
• Partition：Topic 是一个逻辑的概念，它可以细分为多个分区，每个分区只属于单个主题。同一个主题下不同分区包含的消息是不同的，分区在存储层面可以看作一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会分配一个特定的偏移量（offset）。
• Offset：offset 是消息在分区中的唯一标识，Kafka 通过它来保证消息在分区内的顺序性，不过 offset 并不跨越分区，也就是说，Kafka 保证的是分区有序性而不是主题有序性。
• Replication：副本，是 Kafka 保证数据高可用的方式，Kafka 同一 Partition 的数据可以在多 Broker 上存在多个副本，通常只有主副本对外提供读写服务，当主副本所在 broker 崩溃或发生网络一场，Kafka 会在 Controller 的管理下会重新选择新的 Leader 副本对外提供读写服务。
• Record：实际写入 Kafka 中并可以被读取的消息记录。每个 record 包含了 key、value 和 timestamp。

三. Kafka如何保证消息顺序消费

Kafka 在Topic级别本身是无序的，只有partition上才有序，所以为了保证处理顺序，可以自定义分区器，将需顺序处理的数据发送到同一个partition。自定义分区器需要实现接口Partitioner接口并实现 3 个方法:partition,close,configure，在partition方法中返回分区号即可。
Kafka 中发送 1 条消息的时候，可以指定(topic, partition, key)3 个参数，partiton和key是可选的。
Kafka 分布式的单位是partition，同一个partition用一个write ahead log组织，所以可以保证FIFO的顺序。不同partition之间不能保证顺序。因此你可以指定partition，将相应的消息发往同 1个partition，并且在消费端，Kafka 保证1 个partition只能被1 个consumer消费，就可以实现这些消息的顺序消费。
另外，也可以指定 key（比如 order id），具有同 1 个 key 的所有消息，会发往同 1 个partition，那这样也实现了消息的顺序消息。

四. Kafka发送消息选择分区的逻辑

Kafka在数据生产的时候，有一个数据分发策略。默认的情况使用org.apache.kafka.clients.producer.internals.DefaultPartitioner类，这个类中就是定义数据分发的策略。默认策略为：

1. 如果在发消息的时候指定了分区，则消息投递到指定的分区
2. 如果没有指定分区，但是消息的key不为空，则基于key的哈希值来选择一个分区
3. 如果既没有指定分区，且消息的key也是空，则用轮询的方式选择一个分区

五. Kafka如何避免消息丢失

Kafka的消息避免丢失可以从三个方面考虑处理：Producer发送消息避免失败、Broker能成功保存接收到的消息、Consumer确认消费消息。

1. Producer发送消息避免失败
   想要Produce发送消息不失败，那就得知道发送结果，网络抖动这些情况是无法避免的，只能是发送后获取发送结果，那么最直接的方式就是把Kafka默认的异步发送改为同步发送(Broker收到消息后ack回复确认)，这样就能实时知道消息发送的结果，但是这样会让Kafka的发送效率大大降低，因为Kafka在默认的异步发送消息的时候可以批量发送，以此大幅度提高发送效率，因此一般很少使用同步发送的方式，除非消息很重要绝不允许丢失。
   但是我们可以采用添加异步或调函数，监听消息发送的结果，如果失败可以在回调中重试，以此来达到尽可能的发送成功。同时Producer本身提供了一个retries的机制，如果因为网络问题，或者Broker故障 导致发送失败，就是重试。一般这个retries设置3-5次或者更高，同时重试间隔时间也随着次数增长。

2. Broker能成功保存接收到的消息
   Broker要成功的保存接收到的消息并且不丢失，就需要把接收到的消息保存到磁盘。Kafka为了提高性能采用的是异步批量，存储到磁盘的机制，就是有一定的消息量和时间间隔要求的，刷磁盘的这个动作是操作系统来调度的，如果在刷盘之前系统就崩溃了，就会数据丢失。
   针对这个情况，Kafka采用Partition分区ack机制，Partition分区是指一个Topic下的多个分区，有一个Leader分区，其他的都是Follower分区，Leader分区负责接收和被读取消息，Follower分区会通过Replication机制同步Leader的数据，负责高可用(Kafka在2.4之后，Kafka提供了读写分离，Follower也可以提供读取)，当Leader出现故障时会从Follower中选取一个成为新的Leader。那么当一个消息发送到Leader分区之后，Kafka提供了一个acks的参数，Producer可以设置这个参数，去结合broker的Partition机制来共同保障数据的可靠性，这个参数的值有三个

   • 0，表示Producer不需要等待broker的响应，就认为消息发送成功了(可能存在数据丢失)
   • 1，表示Leader收到消息之后,不等待其他的Follower的同步就给Producer发一个确认，如果Leader和Partition挂了就可能存在数据丢失
   • -1，表示Leader收到消息之后还会等待ISR列表(与Leader保持正常连接的Follwer节点列表)中的Follower同步完成，再给Producer返回一个确认，也就是所有分区节点都确认收到消息，保证数据不丢失

3. Consumer确认消费消息
   当Producer确定发送消息成功并且Broker成功保存消息之后，基本上Consumer就肯定能消费到消息。Kafka在消费者消费时有一个offset机制，代表了当前消费者消费到了Partition的哪一条消息。kafka的Consumer的配置中，默认的enable.auto.commit = true,表示在Consumer通过poll方法 获取到消息以后，每过5秒（通过配置项可修改）会自动获取poll中得到的大的offset, 提交给Partition中的offset_consumer(存储 offset 的特定topic)。如果enable.auto.commit = false时，则关闭了自动提交，需要手动的通过应用程序代码进行提交。
   所以在Consumer消费消息时，丢失消息的可能会有两种，比如开启了offset自动提交，但是消息消费失败；或者没有开启自动提交offset，但是在消费消息之前提交了offset。针对这两种情况，可以设置在消息消费完成后手动提交offset。总之Consumer端确认消息消费成功后再提交offset即可保证消息正常消费。

六. Kafka的offset机制

​Kafka中的每个Partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到Partition中。Partition中的每个消息都有一个连续的序号，用于Partition唯一标识一条消息，这个唯一标识就是offset。
Offset从语义上来看拥有两种：Current Offset和Committed Offset。

Current Offset保存在Consumer客户端中，它表示Consumer希望收到的下一条消息的序号。它仅仅在poll()方法中使用。例如，Consumer第一次调用poll()方法后收到了20条消息，那么Current Offset就被设置为20。这样Consumer下一次调用poll()方法时，Kafka就知道应该从序号为21的消息开始读取。这样就能够保证每次Consumer poll消息时，都能够收到不重复的消息。

Committed Offset保存在Broker上，它表示Consumer已经确认消费过的消息的序号。主要通过commitSync和commitAsyncAPI来操作。举个例子，Consumer通过poll()方法收到20条消息后，此时Current Offset就是20，经过一系列的逻辑处理后，并没有调用consumer.commitAsync()或consumer.commitSync()来提交Committed Offset，那么此时Committed Offset依旧是0。
Committed Offset主要用于Consumer Rebalance。在Consumer Rebalance的过程中，一个partition被分配给了一个Consumer，那么这个Consumer该从什么位置开始消费消息呢？答案就是Committed Offset。另外，如果一个Consumer消费了5条消息（poll并且成功commitSync）之后宕机了，重新启动之后它仍然能够从第6条消息开始消费，因为Committed Offset已经被Kafka记录为5。

在Kafka 0.9前，Committed Offset信息保存在zookeeper的consumers/{group}/offsets/{topic}/{partition}目录中（zookeeper并不适合进行大批量的读写操作，尤其是写操作）。而在0.9之后，所有的offset信息都保存在了Broker上的一个名为_consumer_offsets的topic中。

七. Kafka性能高的原因

1. 顺序读写
   Kafka的Partition中写入数据，是通过分段、追加日志的方式，这在很大程度上将读写限制为顺序 I/O（sequential I/O），这在大多数的存储介质上都很快。实际上不管是内存还是磁盘，快或慢关键在于寻址的方式，磁盘分为顺序读写与随机读写，内存也一样分为顺序读写与随机读写。基于磁盘的随机读写确实很慢，但磁盘的顺序读写性能却很高，一般而言要高出磁盘随机读写三个数量级，一些情况下磁盘顺序读写性能甚至要高于内存随机读写。
   

2. Page Cache
   为了优化读写性能，Kafka利用了操作系统本身的Page Cache，就是利用操作系统自身的内存而不是JVM空间内存。这样做可以避免Object消耗，如果是使用 Java 堆，Java对象的内存消耗比较大，通常是所存储数据的两倍甚至更多；还能避免GC问题，随着JVM中数据不断增多，垃圾回收将会变得复杂与缓慢，使用系统缓存就不会存在GC问题。

   相比于使用JVM或in-memory cache等数据结构，利用操作系统的Page Cache更加简单可靠。首先，操作系统层面的缓存利用率会更高，因为存储的都是紧凑的字节结构而不是独立的对象。其次，操作系统本身也对于Page Cache做了大量优化，提供了write-behind、read-ahead以及flush等多种机制。再者，即使服务进程重启，系统缓存依然不会消失，避免了in-process cache重建缓存的过程。

   通过操作系统的Page Cache，Kafka的读写操作基本上是基于内存的，读写速度得到了极大的提升。

3. 零拷贝
   Linux操作系统 零拷贝 机制使用了sendfile方法， 允许操作系统将数据从Page Cache直接发送到网络，只需要最后一步的copy操作将数据复制到 NIC 缓冲区， 这样避免重新复制数据。零拷贝的技术基础是DMA，又称之为直接内存访问。DMA 传输将数据从一个地址空间复制到另外一个地址空间。当CPU 初始化这个传输动作，传输动作本身是由 DMA 控制器来实行和完成。因此通过DMA，硬件则可以绕过CPU，自己去直接访问系统主内存。很多硬件都支持DMA，其中就包括网卡、声卡、磁盘驱动控制器等。通过这种 “零拷贝” 的机制，Page Cache结合sendfile方法，Kafka消费端的性能也大幅提升。这也是为什么有时候消费端在不断消费数据时，我们并没有看到磁盘io比较高，此刻正是操作系统缓存在提供数据。

   当Kafka客户端从服务器读取数据时，如果不使用零拷贝技术，那么大致需要经历这样的一个过程：

   • 操作系统将数据从磁盘上读入到内核空间的读缓冲区中。
   • Kafka应用程序从内核空间的读缓冲区将数据拷贝到用户空间的缓冲区中。
   • Kafka应用程序将数据从用户空间的缓冲区再写回到内核空间的socket缓冲区中。
   • 操作系统将socket缓冲区中的数据拷贝到NIC缓冲区中，然后通过网络发送给客户端。

   如果使用零拷贝技术，那么只需要：

   • 操作系统将数据从磁盘中加载到内核空间的Read Buffer（页缓存区）中
   • 操作系统之间将数据从内核空间的Read Buffer（页缓存区）传输到网卡中，并通过网卡将数据发送给接收方

4. 批量读写
   Kafka数据读写也是批量的而不是单条的。除了利用底层的技术外，Kafka还在应用程序层面提供了一些手段来提升性能，最明显的就是使用批次，在向Kafka写入数据时，可以启用批次写入，这样可以避免在网络上频繁传输单个消息带来的延迟和带宽开销。假设网络带宽为10MB/S，一次性传输10MB的消息比传输1KB的消息10000万次显然要快得多。

5. 批量压缩
   Kafka可以把所有的消息都变成一个批量的文件，并且进行合理的批量压缩，减少网络IO损耗，通过mmap提高I/O速度，写入数据的时候由于单个Partion是末尾添加所以速度最优，读取数据的时候配合sendfile直接输出。并且Kafka支持多种压缩协议，包括Gzip和Snappy压缩协议。

八. Kafka的topic数量太多性能会急剧下降的原因是什么

Kafka的topic数量多性能会下降的主要原因是topic在物理层面以partition为分组，一个topic可以分成若干个partition，partition还可以细分为logSegment，一个partition物理上由多个logSegment组成，logSegment文件由两部分组成，分别为.index文件和.log文件，分别表示为Segment索引文件和数据文件。Kafka在Broker接受并存储消息的时候，是将消息数据使用分段、追加日志的方式写入log文件，在很大程度上将读写限制为顺序 I/O（sequential I/O），那么如果topic数量很多，即使每个topic只有1个partition，也会导致总分区数很多，磁盘读写退化为随机，影响性能。同时Kafka中topic的元数据是在zookeeper中的，大量topic确实会造成性能瓶颈(zk不适合做高并发的读写操作)，不仅在磁盘读写上。而且topic太多造成partition过多。partition是kafka的最小并行单元，每个partition都会在对应的broker上有日志文件。当topic过多，partition增加，日志文件数也随之增加，就需要允许打开更多的文件数。partition过多在controller选举和controller重新选举partition leader的耗时会大大增加，造成kafka不可用的时间延长。

九. Kafka的Replication机制 十. Kafka的Consumer Group 十一. Kafka的零拷贝 十二. Kafka的HW和LEO 十三. Kafka的ISR 十四. Kafka的Rebalance 十五. Kafak为什么依赖zk，zk在其中的作用是什么

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧

分享名称：Kafka的相关知识-创新互联
浏览路径：http://cqcxhl.cn/article/csjgoj.html