一文解密Kafka，Kafka源码设计与实现原理剖析，真正的通俗易懂

Apache Kafka （简称Kafka ）最早是由Linkedln开源出来的分布式消息系统，现在是Apache旗下的一个子项目，并且已经成为开册、领域应用最广泛的消息系统之 Kafka社区也非常活跃，从 版本开始， Kafka 的标语已经从“一个高吞吐量、分布式的消息系统”改为“一个分布式的流平台”

关于Kafka，我打算从入门开始讲起，一直到它的底层实现逻辑个原理以及源码，建议大家花点耐心，从头开始看，相信会对你有所收获。

作为 个流式数据平台，最重要的是要具备下面 个特点

消息系统：

消息系统 也叫作消息队列）主要有两种消息模型：队列和发布订Kafka使用消费组（ consumer group ）统 上面两种消息模型 Kafka使用队列模型时，它可以将处理 作为平均分配给消费组中的消费者成员

下面我们会从 个角度分析Kafka 的几个基本概念，并尝试解决下面 个问题

消息由生产者发布到 fk 集群后，会被消费者消费 消息的消费模型有两种：推送模型（ pu和拉取模型（ pull 基于推送模型的消息系统，由消息代理记录消费者的消费状态 消息代理在将消息推送到消费者后 标记这条消息为已消费

但这种方式无法很好地保证消息的处理语义 比如，消息代理把消息发送出去后，当消费进程挂掉或者由于网络原因没有收到这条消息时，就有可能造成消息丢失（因为消息代理已经 这条消息标记为自己消费了，但实际上这条消息并没有被实际处理） 如果要保证消息的处理语义，消息代理发送完消息后，要设置状态为“已发送”，只有收到消费者的确认请求后才更新为“已消费”，这就需要在消息代理中记录所有消息的消费状态，这种做法也是不可取的

Kafka每个主题的多个分区日志分布式地存储在Kafka集群上，同时为了故障容错，每个分区都会以副本的方式复制到多个消息代理节点上 其中一个节点会作为主副本（ Leader ），其 节点作为备份副本（ Follower ，也叫作从副本）

主副本会负责所有的客户端读写操作，备份副本仅仅从主副本同步数据 当主副本 IH 现在故障时，备份副本中的 副本会被选择为新的主副本 因为每个分区的副本中只有主副本接受读写，所以每个服务端都会作为某些分区的主副本，以及另外一些分区的备份副本这样Kafka集群的所有服务端整体上对客户端是负载均衡的

消息系统通常由生产者「pro ucer 消费者（ co sumer ）和消息代理（ broke 大部分组成，生产者会将消息写入消息代理，消费者会从消息代理中读取消息 对于消息代理而言，生产者和消费者都属于客户端：生产者和消费者会发送客户端请求给服务端，服务端的处理分别是存储消息和获取消息，最后服务端返回响应结果给客户端

新的生产者应用程序使用 af aP oduce 对象代表 个生产者客户端进程 生产者要发送消息，并不是直接发送给 务端 ，而是先在客户端 消息放入队列 然后 一个 息发送线程从队列中消息，以 盐的方式发送消息给服务端 Kafka的记 集器（ Reco dACCUl'lUlato ）负责缓存生产者客户端产生的消息，发送线程（ Sende ）负责读取 集器的批 过网络发送给服务端为了保证客户端 络请求 快速 应， Kafka 用选择器（ Selecto 络连接 读写 理，使网络连接（ Netwo kCl ient ）处理客户端 络请求

追加消息到记录收集器时按照分区进行分组，并放到batches集合中，每个分区的队列都保存了将发送到这个分区对应节点上的 记录，客户端的发送线程可 只使用 Sende 线程迭 batches的每个分区，获取分区对应的主剧本节点，取出分区对应的 列中的批记录就可以发送消息了

消息发送线程有两种消息发送方式 按照分区直接发送 按照分区的目标节点发迭 假设有两台服务器， 题有 个分区，那么每台服务器就有 个分区 ，消息发送线程迭代batches的每个分 接往分区的主副本节点发送消息，总共会有 个请求 所示，我 先按照分区的主副本节点进行分组， 属于同 个节点的所有分区放在一起，总共只有两个请求做法可以大大减少网络的开销

消息系统由生产者 存储系统和消费者组成 章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程 首先我 来了解消费者的 些基础知识

作为分布式的消息系统， Kafka支持多个生产者和多个消费者，生产者可以将消息发布到集群中不同节点的不同分区上；「肖费者也可以消费集群中多个节点的多个分区上的消息 写消息时，多个生产者可以 到同 个分区 读消息时，如果多个消费者同时读取 个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁 同步等方式，在分区级别的日志文件上做些控制

相反，如果约定“同 个分区只可被 个消费者处理”，就不需要加锁同步了，从而可提升消费者的处理能力 而且这也并不违反消息的处理语义：原先需要多个消费者处理，现在交给一个消费者处理也是可以的 3- 给出了 种最简单的消息系统部署模式，生产者的数据源多种多样，它们都统写人Kafka集群 处理消息时有多个消费者分担任务 ，这些消费者的处理逻辑都相同， 每个消费者处理的分区都不会重复

因为分区要被重新分配，分区的所有者都会发生变 ，所以在还没有重新分配分区之前 所有消费者都要停止已有的拉取钱程 同时，分区分配给消费者都会在ZK中记录所有者信息，所以也要先删ZK上的节点数据 只有和分区相关的 所有者 拉取线程都释放了，才可以开始分配分区

如果说在重新分配分区前没有释放这些信息，再平衡后就可能造成同 个分区被多个消费者所有的情况 比如分区Pl 原先归消费者 所有，如果没有释放拉取钱程和ZK节点，再平衡后分区Pl 被分配给消费者 了，这样消费者 和消费者 就共享了分区Pl ，而这显然不符合 fka 中关于“一个分区只能被分配给 个消费者”的限制条件 执行再平衡操作的步骤如下

如果是协调者节点发生故障，服务端会有自己的故障容错机制，选出管理消费组所有消费者的新协调者节，点消费者客户端没有权利做这个工作，它能做的只是等待一段时间，查询服务端是否已经选出了新的协调节点如果消费者查到现在已经有管理协调者的协调节点，就会连接这个新协调节，哉由于这个协调节点是服务端新选出来的，所以每个消费者都应该重新连接协调节点

消费者重新加入消费组，在分配到分区的前后，都会对消费者的拉取工作产生影响 消费者发送“加入组请求”之前要停止拉取消息，在收到“加入组响应”中的分区之后要重新开始拉取消息时，为了能够让客户端应用程序感知消费者管理的分区发生变化，在加入组前后，客户端还可以设置自定义的“消费者再平衡监听器”，以便对分区的变化做出合适的处理

    Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的(partition)、多副本的(replica)，基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

JMS(Java Message Service)是Java提供的一套技术规范

    用来异构系统 集成通信，缓解系统瓶颈，提高系统的伸缩性增强系统用户体验，使得系统模块化和组件化变得可行并更加灵活

(1) 点对点模式(一对一，消费者主动拉取数据，消息收到后消息清除)

    点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。

(2) 发布/订阅模式(一对多，数据生产后，推送给所有订阅者)

    发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。

    kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。

kafka集群支持热扩展

消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

支持数千个客户端同时读写

    一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。

    解耦和生产者和消费者、缓存消息等。

    Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。

    Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

    比如spark streaming和storm

    Kafka每个主题的多个分区日志分布式地存储在Kafka集群上，同时为了故障容错，每个分区都会以副本的方式复制到多个消息代理节点上。其中一个节点会作为主副本(Leader)，其他节点作为备份副本(Follower，也叫作从副本)。主副本会负责所有的客户端读写操作，备份副本仅仅从主副本同步数据。当主副本出现故障时，备份副本中的一个副本会被选择为新的主副本。因为每个分区的副本中只有主副本接受读写，所以每个服务器端都会作为某些分区的主副本，以及另外一些分区的备份副本，这样Kafka集群的所有服务端整体上对客户端是负载均衡的。

    Kafka的生产者和消费者相对于服务器端而言都是客户端。

    Kafka生产者客户端发布消息到服务端的指定主题，会指定消息所属的分区。生产者发布消息时根据消息是否有键，采用不同的分区策略。消息没有键时，通过轮询方式进行客户端负载均衡；消息有键时，根据分区语义（例如hash）确保相同键的消息总是发送到同一分区。

    Kafka的消费者通过订阅主题来消费消息，并且每个消费者都会设置一个消费组名称。因为生产者发布到主题的每一条消息都只会发送给消费者组的一个消费者。所以，如果要实现传统消息系统的“队列”模型，可以让每个消费者都拥有相同的消费组名称，这样消息就会负责均衡到所有的消费者；如果要实现“发布-订阅”模型，则每个消费者的消费者组名称都不相同，这样每条消息就会广播给所有的消费者。

    分区是消费者现场模型的最小并行单位。如下图（图1）所示，生产者发布消息到一台服务器的3个分区时，只有一个消费者消费所有的3个分区。在下图（图2）中，3个分区分布在3台服务器上，同时有3个消费者分别消费不同的分区。假设每个服务器的吞吐量时300MB，在下图（图1）中分摊到每个分区只有100MB，而在下图（图2）中，集群整体的吞吐量有900MB。可以看到，增加服务器节点会提升集群的性能，增加消费者数量会提升处理性能。

    同一个消费组下多个消费者互相协调消费工作，Kafka会将所有的分区平均地分配给所有的消费者实例，这样每个消费者都可以分配到数量均等的分区。Kafka的消费组管理协议会动态地维护消费组的成员列表，当一个新消费者加入消费者组，或者有消费者离开消费组，都会触发再平衡操作。

    Kafka的消费者消费消息时，只保证在一个分区内的消息的完全有序性，并不保证同一个主题汇中多个分区的消息顺序。而且，消费者读取一个分区消息的顺序和生产者写入到这个分区的顺序是一致的。比如，生产者写入“hello”和“Kafka”两条消息到分区P1，则消费者读取到的顺序也一定是“hello”和“Kafka”。如果业务上需要保证所有消息完全一致，只能通过设置一个分区完成，但这种做法的缺点是最多只能有一个消费者进行消费。一般来说，只需要保证每个分区的有序性，再对消息假设键来保证相同键的所有消息落入同一分区，就可以满足绝大多数的应用。

Channel被设计为Event中转临时缓冲区，存储Source收集并且没有被Sink读取的Event，为平衡Source收集和Sink读取数据的速度，可视为Flume内部的消息队列。Channel线程安全并且具有事务性，支持source写失败重复写和sink读失败重复读等操作。

常用的Channel类型有Memory Channel、File Channel、KafkaChannel等。

对比Channel, Memory Channel读写速度快，但是存储数据量小，Flume进程挂掉、服务器停机或者重启都会导致数据丢失。部署Flume Agent的线上服务器内存资源充足、不关心数据丢失的场景下可以使用。

将 event 写入磁盘文件，与 Memory Channel 相比存储容量大，无数据丢失风险。File Channle 数据存储路径可以配置多磁盘文件路径，通过磁盘并行写入提高FileChannel 性能。Flume 将 Event 顺序写入到 File Channel 文件的末尾，在配置文件中通过设置 maxFileSize 参数配置数据文件大小，当被写入的文件大小达到上限时 Flume 会重新创建新的文件存储写入的 Event。当然数据文件数量也不会无限增长，当一个已关闭的只读数据文件中的 Event 被读取完成，并且 Sink 已经提交读取完成的事务，则 Flume 将删除存储该数据的文件。Flume 通过设置检查点和备份检查点实现在 Agent 重启之后快速将 File Channle 中的数据按顺序回放到内存中，保证在 Agent 失败重启后仍然能够快速安全地提供服务。

将Kafka作为Channel存储，Kafka是分布式、可扩展、高容错、高吞吐的分布式系统，Kafka通过优秀的架构设计充分利用磁盘顺序特性，在廉价的硬件条件下完成高效的消息发布和订阅。

Memory Channel在使用的过程中受内存容量的限制不能缓存大量的消息，并且如果Memory Channel中的消息没来得及写入Sink，此时Agent出现故障就会造成数据丢失。File Channel虽然能够缓存更多的消息，但如果缓存下来的消息还没有写入Sink，此时Agent出现故障则File Channel中的消息不能被继续使用，直到该Agent重新恢复才能够继续使用File Channel中的消息。Kafka Channel相对于Memory Channel和File Channel存储容量更大、容错能力更强，弥补了其他两种Channel的短板，如果合理利用Kafka的性能，能够达到事半功倍的效果。

有了Kafka Channel可以在日志收集层只配置Source组件和Kafka Channel组件，不需要再配置Sink组件，减少了日志收集层启动的进程数并且有效降低服务器内存、磁盘等资源使用率，日志汇聚层可以只配置Kafka Channel和Sink，不需要再配置Source，减少日志汇聚层的进程数，这样的配置既能降低服务器的资源使用率又能减少Event在网络之间的传输，有效提高日志采集系统的性能。

Kafka Channel相关操作在orgapacheflumechannelkafka包的KafkaChannel类定义，

kafka相关参数的默认值在orgapachekafkaclientsCommonClientConfigs包中的KafkaChannel-Configuration中。

Kafka的通用配置参数在配置文件中都以“kafka”为前缀，针对Producer或者Consumer的相关配置以“kafkaproducer ”或者“kafkaconsumer ”为前缀，

源码 KafkaChannelConfiguration 中相关默认配置参数定义如下：

说明：agent_name 没有配置Source，只配置了Channel和Sink，使用的Channel类型为Kafka Channel，主题名称为“test_channel”, consumer组id为“test-consumer”, Sink类型为 hdfs 滚动生成文件，对接的Channel为KafkaChannel channel_name。

Kafka 是一个分布式消息队列，具有高性能、持久化、多副本备份、横向扩展能力。生产者往队列里写消息，消费者从队列里取消息进行业务逻辑。一般在架构设计中起到解耦、削峰、异步处理的作用。

（1）生产者和消费者（producer和consumer）：消息的发送者叫 Producer，消息的使用者和接受者是 Consumer，生产者将数据保存到 Kafka 集群中，消费者从中获取消息进行业务的处理。

（2）broker：Kafka 集群中有很多台 Server，其中每一台 Server 都可以存储消息，将每一台 Server 称为一个 kafka 实例，也叫做 broker。

（3）主题（topic）：一个 topic 里保存的是同一类消息，相当于对消息的分类，每个 producer 将消息发送到 kafka 中，都需要指明要存的 topic 是哪个，也就是指明这个消息属于哪一类。

（4）分区（partition）：每个 topic 都可以分成多个 partition，每个 partition 在存储层面是 append log 文件。任何发布到此 partition 的消息都会被直接追加到 log 文件的尾部。为什么要进行分区呢？最根本的原因就是：kafka基于文件进行存储，当文件内容大到一定程度时，很容易达到单个磁盘的上限，因此，采用分区的办法，一个分区对应一个文件，这样就可以将数据分别存储到不同的server上去，另外这样做也可以负载均衡，容纳更多的消费者。

（5）偏移量（Offset）：一个分区对应一个磁盘上的文件，而消息在文件中的位置就称为 offset（偏移量），offset 为一个 long 型数字，它可以唯一标记一条消息。由于kafka 并没有提供其他额外的索引机制来存储 offset，文件只能顺序的读写，所以在kafka中几乎不允许对消息进行“随机读写”。

（6）分布式和分区（distributed、partitioned）

  我们说 kafka 是一个分布式消息系统，所谓的分布式，实际上我们已经大致了解。消息保存在 Topic 中，而为了能够实现大数据的存储，一个 topic 划分为多个分区，每个分区对应一个文件，可以分别存储到不同的机器上，以实现分布式的集群存储。另外，每个 partition 可以有一定的副本，备份到多台机器上，以提高可用性。

总结起来就是：一个 topic 对应的多个 partition 分散存储到集群中的多个 broker 上，存储方式是一个 partition 对应一个文件，每个 broker 负责存储在自己机器上的 partition 中的消息读写。

（7）副本（replicated ）

kafka 还可以配置 partitions 需要备份的个数(replicas),每个 partition 将会被备份到多台机器上,以提高可用性，备份的数量可以通过配置文件指定。

这种冗余备份的方式在分布式系统中是很常见的，那么既然有副本，就涉及到对同一个文件的多个备份如何进行管理和调度。kafka 采取的方案是：每个 partition 选举一个 server 作为“leader”，由 leader 负责所有对该分区的读写，其他 server 作为 follower 只需要简单的与 leader 同步，保持跟进即可。如果原来的 leader 失效，会重新选举由其他的 follower 来成为新的 leader。

至于如何选取 leader，实际上如果我们了解 ZooKeeper，就会发现其实这正是 Zookeeper 所擅长的，Kafka 使用 ZK 在 Broker 中选出一个 Controller，用于 Partition 分配和 Leader 选举。

另外，这里我们可以看到，实际上作为 leader 的 server 承担了该分区所有的读写请求，因此其压力是比较大的，从整体考虑，有多少个 partition 就意味着会有多少个leader，kafka 会将 leader 分散到不同的 broker 上，确保整体的负载均衡。

Apache Kafka 的一个关键依赖是 Apache Zookeeper，它是一个分布式配置和同步服务。Zookeeper 是 Kafka 代理和消费者之间的协调接口。Kafka 服务器通过 Zookeeper 集群共享信息。Kafka 在 Zookeeper 中存储基本元数据，例如关于主题，代理，消费者偏移(队列读取器)等的信息。

由于所有关键信息存储在 Zookeeper 中，并且它通常在其整体上复制此数据，因此Kafka代理/ Zookeeper 的故障不会影响 Kafka 集群的状态。Kafka 将恢复状态，一旦 Zookeeper 重新启动。 这为Kafka带来了零停机时间。Kafka 代理之间的领导者选举也通过使用 Zookeeper 在领导者失败的情况下完成。

以上流程将重复，直到消费者停止请求。

消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。

在队列消息传递系统而不是单个消费者中，具有相同组 ID 的一组消费者将订阅主题。 简单来说，订阅具有相同 Group ID 的主题的消费者被认为是单个组，并且消息在它们之间共享。 让我们检查这个系统的实际工作流程。

此功能也称为使用者组。 同样，Kafka 将以非常简单和高效的方式提供两个系统中最好的。

https://wwworchomecom/22

xxindex ：相对offset ,绝对position

xxlog :offset,position,message

xxtimeindex:time,相对offset

(1)查找segment file

00000000000000000000index表示最开始的文件，起始偏移量(offset)为0第二个文件00000000000000368769index的消息量起始偏移量为368770 = 368769 + 1同样，第三个文件00000000000000737337index的起始偏移量为737338=737337 + 1，其他后续文件依次类推，以起始偏移量命名并排序这些文件，只要根据offset 二分查找文件列表，就可以快速定位到具体文件。

当offset=368776时定位到00000000000000368769index|log

(2)通过segment file查找message

通过第一步定位到segment file，当offset=368776时，依次定位到00000000000000368769index的元数据物理位置和00000000000000368769log的物理偏移地址，然后再通过00000000000000368769log顺序查找直到offset=368776为止。

https://blogcsdnnet/hyj_king/article/details/105710993

https://wwworchomecom/28

https://wwworchomecom/29

问题

https://wwworchomecom/20

发送可靠性： 发送消息后，等待确认（需要确保 足够副本节点可用状态）

提交offset，但处理消息失败，需要保存offset，重复消费

重复消费：有业务端 来保障（比如数据表唯一性）

https://wwworchomecom/22