Kafka 我以为我对Kafka很了解，直到我看了这篇文章

"

卡夫卡是一个消息系统，最初是从领英开发的，用作领英活动流和运营数据处理管道的基础。

现在，它已经被许多不同的公司用作各种数据管道和消息系统。活动流数据是几乎所有网站在报告其网站使用情况时使用的最常见的数据部分。

活动数据包括页面视图、查看内容信息、搜索情况等。

通常处理这类数据的方法是将各种活动以日志的形式写入某个文件，然后定期对这些文件进行统计分析。

运营数据是指服务器的性能数据(CPU、IO利用率、请求时间、服务日志等)。运营数据的统计方法有很多。

近年来，活动和运营数据处理已经成为网站软件产品功能的重要组成部分，这需要稍微复杂一点的基础设施来支持。

卡夫卡基本概念

卡夫卡是一个基于发布/订阅的分布式消息系统，其主要设计目标如下:

以时间复杂度为 O(1) 的方式提供消息持久化能力，即使对 TB 级以上数据也能保证常数时间复杂度的访问性能。高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒 100K 条以上消息的传输。支持 Kafka Server 间的消息分区，及分布式消费，同时保证每个 Partition 内的消息顺序传输。同时支持离线数据处理和实时数据处理。Scale out：支持在线水平扩展。

生产者和消费者

卡夫卡有两种基本类型的客户:

生产者（Producer）消费者（Consumer）

另外还有数据集成的卡夫卡Connect API、流媒体处理的卡夫卡Streams等高端客户端，但这些高端客户端的底层还是生产者和消费者API，只是封装在上层。

很容易理解，生产者(也称为发布者)创建消息，而消费者(也称为订阅者)负责消费或阅读消息。

一个主题和一个分区

在卡夫卡，消息是按主题分类的，每个主题对应一个“消息队列”，有点类似于数据库中的一个表。

但是，如果把同类的消息都塞进一个“中心”队列，势必缺乏可扩展性。无论生产者/消费者的数量增加还是消息的数量增加，系统的性能或存储都可能耗尽。

我们用一个生活实例来说明，A市生产的一种商品需要通过公路运输到B市。

那么单通道高速公路就会出现“吞吐量不足”的问题，无论是“A市商品增加”还是“现在C市到B市的东西运输”。

因此，我们现在引入了分区的概念，它以类似于“允许更多磁道”的方式横向扩展了我们的主题。

代理和集群(集群)

一个卡夫卡服务器，也叫Broker，接收制作人发来的消息，存储在磁盘上；同时，代理服务于消费者的请求，提取分区消息并返回到目前为止已经提交的消息。

使用特定的机器硬件，代理每秒可以处理数千个分区和数百万条消息。(现在在百万量级。我去查看了一下。看来集群条件下吞吐量还是挺高的。)

几个代理组成一个集群，集群中的一个代理成为集群控制器，负责管理集群，包括为代理分配分区、监控代理故障等。

在集群中，一个分区由一个代理负责，代理也被称为该分区的领导者。

当然，一个分区可以被复制到多个代理，以实现冗余，这样当代理出现故障时，它的分区可以被重新分配给其他代理来负责。

下图是一个示例:

卡夫卡的一个关键属性是日志保留。我们可以配置主题的消息保留策略，例如只保留特定时间段的日志，或者只保留特定大小的日志。

当超过这些限制时，旧邮件将被删除。我们还可以为某个主题单独设置消息过期策略，可以针对不同的应用进行个性化设置。

多个集群

随着业务的发展，我们经常需要多个集群，通常有以下原因:

基于数据的隔离基于安全的隔离多数据中心（容灾）

建立多个数据中心时，往往需要实现消息互通。例如，如果用户修改了个人数据，无论哪个数据中心处理后续请求，都需要反映更新。或者，来自多个数据中心的数据需要聚合到一个中央控制中心进行数据分析。

上述分区复制的冗余机制只适用于同一个卡夫卡集群，卡夫卡提供的MirrorMaker工具可以用于多个卡夫卡集群的消息同步。

本质上，MirrorMaker只是一个卡夫卡式的消费者和生产者，由一个队列连接起来。它消耗来自一个集群的消息，然后向另一个集群产生消息。

卡夫卡的设计与实现

以上，我们知道卡夫卡的一些基本概念，但是作为一个成熟的“消息队列”中间件，有很多有趣的设计值得我们考虑。下面简单列举一些。

卡夫卡存储在文件系统中

是的，你首先应该知道文件系统上存在卡夫卡的消息。卡夫卡非常依赖文件系统来存储和缓存消息。人们普遍认为“磁盘慢”，所以对这种设计持怀疑态度。

事实上，磁盘比预期的快得多，慢得多，这取决于它们的使用方式；一个好的磁盘结构设计可以让它和网速一样快。

现代操作系统已经为磁盘读写制定了一些优化方案，以加快磁盘访问速度。

例如，预读会提前将相对较大的磁盘快速读入内存。写后将许多小的逻辑写操作组合成一个大的物理写操作。

此外，操作系统会将主内存的所有剩余空备用内存空作为磁盘缓存，所有磁盘读写操作都会通过统一磁盘缓存(除了直接I/O会绕过磁盘缓存)。

根据这些优化特性，如果是顺序访问磁盘，在某些情况下可能比随机内存访问快，甚至和网络速度差不多。

上述主题实际上是一个逻辑概念，面向消费者和生产者，物理存储分区，每个分区最终对应一个目录，所有消息和索引文件都存储在这个目录中。

默认情况下，创建主题时，如果未指定分区数量，将只创建一个分区。

例如，如果我创建了一个名为test的主题，并且没有指定Partition的数量，我将默认创建一个test-0的文件夹。这里的命名规则是:

发布到分区的任何消息都会被追加到分区数据文件的尾部，这使得卡夫卡的效率非常高(验证了顺序磁盘写入的效率高于随机内存写入，这是卡夫卡高吞吐量的重要保证)。

当每条消息发送到代理时，它将根据分区规则选择存储哪个分区。如果分区规则设置合理，所有消息可以均匀分布在不同的分区中。

卡夫卡的底层储物设计

假设卡夫卡集群中只有一个Broker，我们创建了两个名为“Topic1”和“Topic2”的主题，分区号分别为1和2。

然后将在我们的根目录下创建以下三个文件夹:

| -主题1-0

| -主题2-0

| -主题2-1

在卡夫卡的文件存储中，同一个主题下有许多不同的分区，每个分区都是一个目录。

每个目录均匀分布在多个大小相等的段文件中，段文件由索引文件和数据文件组成，它们总是成对出现，后缀为”。索引“和”。日志"/>

其中元数据

请注意，索引文件不是从0开始的，也不是每次都递增1。这是因为卡夫卡采用稀疏索引存储方式，每隔某个字节的数据就建立一个索引。

它减小了Index文件的大小，使索引映射到内存成为可能，减少了查询时的磁盘IO开销，并且不会给查询带来太多的时间消耗。

因为它的文件名是前一个段的最后一条消息的偏移量，所以当您需要查找指定偏移量的消息时，您可以在所有段的文件名中找到它所属的二分搜索法段。

然后在它的索引文件中找到它对应的文件上的物理位置，就可以取出Message了。

因为消息在Partition的Segment数据文件中是按顺序读写的，消息消费后不会被删除(删除策略是针对过期的Segment文件)，这也是顺序磁盘IO存储设计者Kafka性能高的原因。

卡夫卡如何准确知道《消息》的偏移量？这是因为标准数据存储结构是在卡夫卡中定义的，分区中的每个消息包含以下三个属性:

Offset：表示 Message 在当前 Partition 中的偏移量，是一个逻辑上的值，唯一确定了 Partition 中的一条 Message，可以简单的认为是一个 ID。MessageSize：表示 Message 内容 Data 的大小。Data：Message 的具体内容。

生产者设计概要

发消息之前，先问几个问题:是不是每条消息都很关键，不能容忍损失？偶尔重复留言可以吗？我们关心的是消息延迟还是写消息的吞吐量？

比如有一个信用卡交易处理系统，在交易发生时向卡夫卡发送消息，另一个服务根据规则引擎读取消息并检查交易是否通过，并通过卡夫卡返回结果。

对于这样的服务，消息既不能丢失，也不能重复。因为交易量大，所以吞吐量需要尽可能大，延迟可以稍微高一些。

再比如，如果我们需要收集用户在网页上的点击数据，对于这样的场景，少量的消息丢失或重复是可以容忍的，延迟多长时间也没关系，只要不影响用户体验，吞吐量是根据实时用户数来决定的。

不同的服务需要使用不同的编写方法和配置。具体方式这里就不讨论了。现在让我们看看生产者写消息的基本过程:

流程如下:

首先，我们需要创建一个 ProducerRecord，这个对象需要包含消息的主题（Topic）和值（Value），可以选择性指定一个键值（Key）或者分区（Partition）。发送消息时，生产者会对键值和值序列化成字节数组，然后发送到分配器（Partitioner）。如果我们指定了分区，那么分配器返回该分区即可；否则，分配器将会基于键值来选择一个分区并返回。选择完分区后，生产者知道了消息所属的主题和分区，它将这条记录添加到相同主题和分区的批量消息中，另一个线程负责发送这些批量消息到对应的Kafka Broker。当 Broker 接收到消息后，如果成功写入则返回一个包含消息的主题、分区及位移的 RecordMetadata 对象，否则返回异常。生产者接收到结果后，对于异常可能会进行重试。

消费者设计总结

①消费者和消费群体

假设这个场景:我们从卡夫卡那里读取消息，检查它们，最后产生结果数据。

我们可以创建一个消费者实例来实现这一点，但是如果生产者写消息的速度比消费者读消息的速度快呢？

随着时间的推移，消息积累越来越严重。对于这个场景，我们需要添加更多的消费者来横向扩展。

卡夫卡的消费者是一个消费群体的一部分。当多个消费者组成一个消费者组来消费一个主题时，每个消费者将从不同的分区接收消息。

假设有一个T1话题，有四个分区。同时，我们有一个消费群体G1，它只有一个消费者C1。

然后，消费者C1将从这四个分区接收消息，如下所示:

如果我们向消费者组G1添加一个新的消费者C2，每个消费者将分别从两个分区接收消息，如下所示:

如果增加到4个消费者，那么每个消费者都会收到一条划分的消息，如下所示:

但是，如果我们继续将消费者添加到该消费者组，剩余的消费者将空闲置，并且不会收到任何消息:

综上所述，我们可以通过增加消费群体中的消费者数量来横向扩张，从而增强消费能力。

这也是为什么在创建主题时建议使用更多的分区，以增加消费者，提高高消费负载下的性能。

另外，消费者的数量不要超过分区的数量，因为多余的消费者在没有任何帮助的情况下是空空闲的。

卡夫卡的一个非常重要的特点就是只需要写一次消息，就可以支持任意数量的应用读取消息。

换句话说，每个应用程序都可以读取大量的消息。为了让每一个应用程序都能读取到大量的消息，应用程序需要有不同的消费群。

对于上面的例子，如果我们添加一个新的消费者组G2，并且这个消费者组有两个消费者，那么它将是这样的:

在这种情况下，消费者组G1和消费者组G2都可以接收T1主题的完整消息，并且它们在逻辑上属于不同的应用程序。

最后总结一下，如果应用需要阅读满量的消息，请为应用设置消费群；如果应用的消费能力不足，那么可以考虑在这个消费群体中增加消费者。

②消费群体与分区之间的再平衡

可以看出，当一个新的消费者加入消费者组时，它将消费一个或多个分区，这些分区以前是由其他消费者负责的。

另外，当消费者离开消费群时(比如重启、停机等。)，它消耗的分区会分配给其他分区。

这种现象叫做再平衡。再平衡是卡夫卡的一个重要属性，它保证了高可用性和横向扩展。

但是，还需要注意的是，在重新平衡期间，所有消费者都不能消费消息，从而导致整个消费者组暂时不可用。

而且重新平衡分区还会导致原来的消费者状态过期，导致消费者需要再次更新状态，这也会降低这段时间的消费性能。后面我们会讨论如何安全平衡，如何尽可能避免。

消费者通过定期向作为组协调者的经纪人发送心跳来保持在消费者组中的活力。

该代理不是固定的，每个消费者群体可能不同。当消费者拉一条消息或提交它时，它发送一个心跳。

如果某个使用者超过一定时间未能发送心跳，其会话将过期，组协调器将认为该使用者已停机，然后触发重新平衡。

可以看出，从消费者的停机时间到会话到期有一段时间，在这段时间内，消费者的分区不能消费消息。

通常，我们可以优雅地关闭，这样消费者就会向组协调器发送离开的消息，这样组协调器就可以立即重新平衡，而不必等待会话到期。

在0.10.1版中，卡夫卡修改了心跳机制，将发送心跳和拉消息分开，使得发送心跳频率不受拉频率的影响。

此外，更高版本的卡夫卡支持配置消费者在不拉消息的情况下可以存活多长时间。这种配置可以避免活锁。动态锁定意味着应用程序没有故障，但由于某种原因无法进一步使用。

③分区和消费模式

如上所述，卡夫卡的一个主题中的消息被分布并存储在多个分区中。消费时，消费组需要从不同的分区获取消息，那么如何最终重构Topic中消息的顺序呢？

答案是:没有办法。卡夫卡只保证消息在分区内有序，不考虑全局情况。

下一个问题是:分区中的消息可以被(不同的消费组)多次消费，那么什么时候删除分区中消费的消息呢？Partition如何知道一个消费群当前的消费位置？

分区从不删除消息，除非它过期，不管它是否被使用。例如，如果保留时间设置为2天，任何组都可以在2天内使用该邮件，并且该邮件将在2天后自动删除。

分区将为每个消费组保存一个偏移量，并记录该组消费的位置。如下图:

④为什么卡夫卡是拉模式

消费者应该向Broker索要数据(Pull)还是Broker向消费者推送数据？

作为一个消息系统，卡夫卡遵循传统的方式，选择生产者向经纪人推送消息，消费者从经纪人那里拉消息。

一些以日志为中心的系统，比如Facebook的Scribe和Cloudera的Flume，都采用了Push模式。其实Push模式和Pull模式各有利弊。

推送模式很难适应不同消费率的消费者，因为消息发送速率是由Broker决定的。

推送模式的目标是尽可能快地传递消息，但消费者很容易没有时间处理消息，其特点是拒绝服务和网络拥塞。

而拉模式可以根据消费者的消费能力以适当的速率消费消息。

对于卡夫卡来说，Pull模式更适合。拉模式可以简化代理的设计，消费者可以独立控制消费消息的速率。

同时，消费者可以控制自己的消费方式，即可以批量消费，也可以逐个消费，同时可以选择不同的提交方式，实现不同的传输语义。

卡夫卡如何保证可靠性

当我们讨论可靠性时，我们总是提到保证*这个词。可靠性保证是基础，我们基于这些基础构建我们的应用程序。

比如关系数据库的可靠性保证是ACID，意思是原子性、一致性、隔离性、持久性。

卡夫卡的可靠性保证有以下四点:

对于一个分区来说，它的消息是有序的。如果一个生产者向一个分区先写入消息 A，然后写入消息 B，那么消费者会先读取消息 A 再读取消息 B。当消息写入所有 in-sync 状态的副本后，消息才会认为已提交（committed）。 这里的写入有可能只是写入到文件系统的缓存，不一定刷新到磁盘。生产者可以等待不同时机的确认，比如等待分区主副本写入即返回，后者等待所有 in-sync 状态副本写入才返回。一旦消息已提交，那么只要有一个副本存活，数据不会丢失。消费者只能读取到已提交的消息。

利用这些基本保证，我们建立了一个可靠的系统。这时候我们需要考虑一个问题:我们的应用需要多大的可靠性？

可靠性不是免费的，它与系统可用性、吞吐量、延迟和硬件价格密切相关。所以我们往往需要做出取舍，一味追求可靠性是不现实的。

开始建造卡夫卡

通过上面的描述，我们对卡夫卡是谁有了一个大致的了解，现在我们正在努力自己构建一个来实际体验。

第一步:下载卡夫卡

这里以Mac OS为例，在安装了Homebrew的情况下执行以下代码:

brewinstall kafka

由于卡夫卡依赖Zookeeper，下载时会自动下载。

步骤2:启动服务

我们需要将卡夫卡的监听地址和端口修改为本地主机:9092:

VI/usr/local/etc/Kafka/server . properties

然后将其修改为如下图所示:

依次启动动物园管理员和卡夫卡:

brew services startzookeeper

brew services startkafka

然后执行以下语句创建一个名为“测试”的主题:

Kafka-topics-create-zoo keeper localhost:2181-replication-factor 1-partitions 1-topic test

我们可以使用以下命令查看主题列表:

Kafka-topics-list-zoo keeper localhost:2181

第三步:发送信息

然后，我们创建一个新的控制台，并运行以下命令来创建一个消费者关注的主题:

kafka-控制台-消费者-引导-服务器localhost: 9092 -主题测试-从头开始

使用控制台向刚刚创建的主题添加消息，并观察刚刚创建的使用者窗口:

Kafka-console-producer-broker-list localhost:9092-主题测试

通过消费者窗口可以看到正确的信息:

参考文献:

Kafka 设计解析（一）：Kafka 背景及架构介绍Kafka系列（一）初识KafkaKafka 入门介绍Kafka 中的 Topic 为什么要进行分区? - 知乎Kafka 的设计与实践思考Kafka系列（六）可靠的数据传输

来源:转载自微信微信官方账号:我没有三颗心(ID: wmyskXZ)