Kafka 技术汇总

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

AWS在re:Invent 2018大会上首先发布了托管Apache Kafka消息队列服务（Amazon Managed Streaming for Apache Kafka，MSK）的消息，现在已经从预览成为正式服务。

Apache Kafka是一个分布式的消息队列系统，其使用发布以及订阅的架构，将产生的流数据的应用与利用流数据的角色分离。Apache Kafka让使用者可以捕捉如消息队列事件、交易、物联网等事件，或是应用与日志等流数据，还能实时进行分析，连续不间断地转换数据，并再将收到的数据经过处理后，分发到其他的数据湖和数据库中。

AWS提到，用户在生产环境中要配置Apache Kafka，需要克服一些障碍，特别是在后续的管理以及规模扩展工作上，而现在AWS正式推出的MSK服务，则由AWS负责管理任务，让用户可以简单地配置使用，而且由于近几个版本的Kafka，都需要与节点协调程序Zookeeper共同使用，因此MSK服务也只要简单地设定，就能让Kafka与ZooKeeper一同运行。

使用MSK服务，用户可以在几分钟内创建集群，并使用AWS身分管理与访问控制IAM管理集群操作，也能通过ACM（AWS Certificate Manager）完全托管的TLS私密凭证颁发机构授权客户端，以TLS加密数据，并使用KMS（AWS Key Management Service）中的密钥加密其他数据。当服务器发生故障时，MSK还会替换故障机器，自动执行修补，用户可以从Amazon CloudWatch中，监控服务的状态指标。

AWS表示，MSK与Kafka 111和210版本完全兼容，因此用户可以在AWS直接执行原本的Kafka应用以及工具，而不需要修改任何的代码，用户能使用开源工具MirrorMaker，将数据从现有的Kafka集群直接迁移到MSK上。

MSK的计价方式是以Kafka Broker以及配置存储每小时计价，MSK的数据传输费用与原本的AWS数据传输相同，而集群所使用的Zookeeper节点，还有区域集群的Broker和Zookeeper节点互传数据是不额外收费的。现在用户已经可以在大部分的AWS区域使用到MSK服务，包括北美、亚洲与欧洲。

Producer： 生产者，发送消息的一方。生产者负责创建消息，然后将其发送到 Kafka 服务器上。

Consumer： 消费者，接受消息的一方。消费者连接到 Kafka 服务器上并接收消息，进而进行相应的业务逻辑处理。

Consumer Group： 消费者集合，一个消费者组可以包含一个或者多个消费者。使用多分区 + 多消费者的方式，可以极大提高下游系统处理速度。同一消费者组中的消费者不会重复消费消息，不同的消费者组之间不会互相影响，都能收到全部消息。kafka就是通过消费组来实现P2P模式和广播模式的。

Broker： Kafka 服务器。

Topic： Kafka中的消息维度，一个Topic类似一个queue。生产者将消息发送到特定的Topic，消费者通过Topic进行消费。

Partition： 分区，分区是属于Topic逻辑概念下的一个分区，每个分区只属于一个Topic，一个Topic通常有多个分区，每个分区包含的消息是不同的，分区在存储层面可以看做一个可追加的日志文件，消息在被追加到分区日志文件时，会分配一个特定的便宜了（offset）。

Offset： 分区中的消息的唯一标识，用它来保证消息在分区内的顺序性，offset不跨分区，也就是说，Kafka保证消息在分区内的有序性，不保证消息在Topic下的有序性

Replication： 副本，是Kafka保证数据高可用的方式。同一Partition的数据可以在多个Broker（kafka服务器）上存在多个副本，通常只有主副本提供读写服务，当主副本发生故障，Kafka会在Controller的管理下，选择新的副本作为主副本提供读写服务

Follower： 从副本，相对于主副本，从副本只同步主副本数据，不提供读写服务。

Record： 写入kafka中的消息，每个消息包含了key、value和timestamp。

生产者-消费者是一种设计模式，是在生产者和消费者之间添加一个中间件来达到解耦的目的。

Zookeeper是一个成熟的分布式协调服务，它可以为分布式服务提供分布式配置服务、同步服务和命名注册等能力。任何分布式服务都需要一种协调任务的方法，Kafka使用Zookeeper来进行任务协调，也有一些其他技术具有自己的内置任务协调机制。

Kafka将Broker、Topic和Partitin的元数据存储在Zookeeper上。

Kafka使用Zookeeper完成以下功能：

Controller是从Broker中选举出来的，负责分区 Leader 和 Follower 的管理。当某个分区的 leader 副本发生变化，由Controller负责为该分区选举新的 leader 副本。当某个分区的同步副本集合发生变化时，由Controller负责通知所有Broker更新元数据信息。

Controller的选举依赖Zookeeper，成功竞选为控制器的Broker会在Zookeeper中创建一个/controller临时节点。

选举过程： Broker首先尝试读取/controller节点中的brokerid值，如果brokerid值不为-1，表示已经存在Broker当选Controller，否则尝试创建/controller节点，创建成功后将当前brokerid写入/controller节点，作为 activeControllerId

主要职责： controller选举出来作为整个Broker集群的管理者，管理所有集群信息和元数据。

Kafka 的网络通信模型是基于 NIO 的Reactor 多线程模型来设计的。其中包含一个Acceptor线程用于处理连接，多个 Processor 线程 select 和 read socket 请求，一个Processor 由包含多个 Handler 线程处理请求并响应。

顺序写：

零拷贝：

PageCache： producer 生成消息到 Broker 时，Broker 会使用 pwrite() 系统调用，按偏移量写入数据。写入时，会先写入 page cache。Consumer 消费消息时，Broker会使用sendfile() 系统调用，零拷贝的将数据从 page cache 传输到 Broker 的 Socket Buffer，通过网络传输。因此当Kafka的生产速率和消费速率相差不大时，就能几乎只靠 page cache 的读写完成整个生产-消费过程，磁盘访问非常少

网络模型： Kafka基于NIO，采用Reactor线程模型，实现了自己的RPC通信。 一个Acceptor线程处理新的连接，多个Processor线程select 和 read socket请求，多个Handler线程处理请求并响应（I/O多路复用）。

批量与压缩： Kafka Producer 向 Broker 发送消息不是一条一条发送，而是按批发送。且roducer、Broker 和 Consumer 使用相同的压缩算法，在 producer 向 Broker 写入数据，Consumer 向 Broker 读取数据时甚至可以不用解压缩，最终在 Consumer Poll 到消息时才解压，这样节省了大量的网络和磁盘开销。

分区并发： Kafka 的 Topic 可以分成多个 Partition，每个 Paritition 类似于一个队列，保证数据有序。同一个 Group 下的不同 Consumer 并发消费 Paritition，分区实际上是调优 Kafka 并行度的最小单元，因此，可以说，每增加一个 Paritition 就增加了一个消费并发。

文件结构：

Kafka 消息是以 Topic 为单位进行归类，各个 Topic 之间是彼此独立的，互不影响。每个 Topic 又可以分为一个或多个分区。每个分区各自存在一个记录消息数据的日志文件。

Kafka 每个分区日志在物理上实际按大小被分成多个 Segment。

index 采用稀疏索引，这样每个 index 文件大小有限，Kafka 采用mmap的方式，直接将 index 文件映射到内存，这样对 index 的操作就不需要操作磁盘 IO。

Kafka 充分利用二分法来查找对应 offset 的消息位置

和其他消息队列相比,Kafka的优势在哪里？

队列模型了解吗？Kafka 的消息模型知道吗？

Kafka 如何保证消息不重复消费？

kafka出现消息重复消费的原因：

解决方案：

参考1： Kafka性能篇：为何Kafka这么"快"？

参考2： Kafka原理篇：图解kakfa架构原理

Kafka到底是个啥？用来干嘛的？

官方定义如下：

翻译过来，大致的意思就是，这是一个实时数据处理系统，可以横向扩展，并高可靠！

实时数据处理 ，从名字上看，很好理解，就是将数据进行实时处理，在现在流行的微服务开发中，最常用实时数据处理平台有 RabbitMQ、RocketMQ 等消息中间件。

这些中间件，最大的特点主要有两个：

在早期的 web 应用程序开发中，当请求量突然上来了时候，我们会将要处理的数据推送到一个队列通道中，然后另起一个线程来不断轮训拉取队列中的数据，从而加快程序的运行效率。

但是随着请求量不断的增大，并且队列通道的数据一致处于高负载，在这种情况下，应用程序的内存占用率会非常高，稍有不慎，会出现内存不足，造成程序内存溢出，从而导致服务不可用。

随着业务量的不断扩张，在一个应用程序内，使用这种模式已然无法满足需求，因此之后，就诞生了各种消息中间件，例如 ActiveMQ、RabbitMQ、RocketMQ等中间件。

采用这种模型，本质就是将要推送的数据，不在存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。

消息中间件 ：主要的职责就是保证能接受到消息，并将消息存储到磁盘，即使其他服务都挂了，数据也不会丢失，同时还可以对数据消费情况做好监控工作。

应用程序 ：只需要将消息推送到消息中间件，然后启用一个线程来不断从消息中间件中拉取数据，进行消费确认即可！

引入消息中间件之后，整个服务开发会变得更加简单，各负其责。

Kafka 本质其实也是消息中间件的一种，Kafka 出自于 LinkedIn 公司，与 2010 年开源到 github。

LinkedIn 的开发团队，为了解决数据管道问题，起初采用了 ActiveMQ 来进行数据交换，大约是在 2010 年前后，那时的 ActiveMQ 还远远无法满足 LinkedIn 对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，LinkedIn 决定研发自己的消息传递系统， Kafka 由此诞生 。

在 LinkedIn 公司，Kafka 可以有效地处理每天数十亿条消息的指标和用户活动跟踪，其强大的处理能力，已经被业界所认可，并成为大数据流水线的首选技术。

先来看一张图， 下面这张图就是 kafka 生产与消费的核心架构模型 ！

如果你看不懂这些概念没关系，我会带着大家一起梳理一遍！

简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：

与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个分区 Partition 的概念。

这个分区的意思就是说，如果你创建的 topic 有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。

这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！

这是 kafka 与其他的消息系统最大的不同！

和其他的中间件一样，kafka 每次发送数据都是向 Leader 分区发送数据，并顺序写入到磁盘，然后 Leader 分区会将数据同步到各个从分区 Follower ，即使主分区挂了，也不会影响服务的正常运行。

那 kafka 是如何将数据写入到对应的分区呢？kafka中有以下几个原则：

与生产者一样，消费者主动的去kafka集群拉取消息时，也是从 Leader 分区去拉取数据。

这里我们需要重点了解一个名词： 消费组 ！

考虑到多个消费者的场景，kafka 在设计的时候，可以由多个消费者组成一个消费组，同一个消费组者的消费者可以消费同一个 topic 下不同分区的数据，同一个分区只会被一个消费组内的某个消费者所消费，防止出现重复消费的问题！

但是不同的组，可以消费同一个分区的数据！

你可以这样理解，一个消费组就是一个客户端，一个客户端可以由很多个消费者组成，以便加快消息的消费能力。

但是，如果一个组下的消费者数量大于分区数量，就会出现很多的消费者闲置。

如果分区数量大于一个组下的消费者数量，会出现一个消费者负责多个分区的消费，会出现消费性能不均衡的情况。

因此，在实际的应用中，建议消费者组的 consumer 的数量与 partition 的数量保持一致！

光说理论可没用，下面我们就以 centos7 为例，介绍一下 kafka 的安装和使用。

kafka 需要 zookeeper 来保存服务实例的元信息，因此在安装 kafka 之前，我们需要先安装 zookeeper。

zookeeper 安装环境依赖于 jdk，因此我们需要事先安装 jdk

下载zookeeper，并解压文件包

创建数据、日志目录

配置zookeeper

重新配置 dataDir 和 dataLogDir 的存储路径

最后，启动 Zookeeper 服务

到官网 http://kafkaapacheorg/downloadshtml 下载想要的版本，我这里下载是最新稳定版 280 。

按需修改配置文件 serverproperties （可选）

serverproperties 文件内容如下：

其中有四个重要的参数：

可根据自己需求修改对应的配置！

启动 kafka 服务

创建一个名为 testTopic 的主题，它只包含一个分区，只有一个副本：

运行 list topic 命令，可以看到该主题。

输出内容：

Kafka 附带一个命令行客户端，它将从文件或标准输入中获取输入，并将其作为消息发送到 Kafka 集群。默认情况下，每行将作为单独的消息发送。

运行生产者，然后在控制台中键入一些消息以发送到服务器。

输入两条内容并回车：

Kafka 还有一个命令行使用者，它会将消息转储到标准输出。

输出结果如下：

本文主要围绕 kafka 的架构模型和安装环境做了一些初步的介绍，难免会有理解不对的地方，欢迎网友批评、吐槽。

由于篇幅原因，会在下期文章中详细介绍 java 环境下 kafka 应用场景！

详细安装访问： https://wwwjianshucom/p/c74e0ec577b0

macOS 可以用homebrew快速安装,访问地址: https://wwwjianshucom/p/cddd25da8061

原文链接： https://wwwjianshucom/p/06884c5bf3f1

查看topic列表：

创建topic：

--create ：创建命令；

--topic ：后面指定topic名称；

--replication-factor ：后面指定副本数；

--partitions ：指定分区数，根据broker的数量决定；

--zookeeper ：后面指定 zookeeperconnect 的zk链接

查看某个topic：

Kafka 作为消息系统的一种， 当然可 以像其他消 息中 间件一样作为消息数据中转的平台。 下面以 Java 语言为例，看一下如何使用 Kafka 来发送和接收消息。

1、引入依赖

2、消息生产者

示例 中用 KafkaProducer 类来创建一个消息生产者,该类的构造函数入参是一系列属性值。下面看一下这些属性具体都是什么含义。

bootstrapservers 表示 Kafka 集群 。 如果集群中有多台物理服务器，则服务器地址之间用逗号分隔， 比如” 19216811 :9092,19216812:9092” 。 localhost 是笔者电脑的地址，9092 是 Kafka 服务器默认监听的端口号。

keyserializer 和 valueserializer 表示消息的序列化类型 。 Kafka 的消息是以键值对的形式发送到 Kafka 服务器的，在消息被发送到服务器之前，消息生产者需要把不同类型的 消息序列化为 二 进制类型，示例中是发送文本消息到服务器 ， 所以使用的是StringSerializer。

keydeserializer 和 valuedeserializer 表示消息的反序列化类型。把来自 Kafka 集群的二进制消 息反序列 化 为指定 的 类型,因为序列化用的是String类型，所以用StringDeserializer 来反序列化。

zkconnect 用于指定 Kafka 连接 ZooKeeper 的 URL ，提供了基于 ZooKeeper 的集群服务器自动感知功能， 可以动态从 ZooKeeper 中读取 Kafka 集群配置信息。

有 了 消息生产者之后 ， 就可以调用 send 方法发送消息了。该方法的入参是 ProducerRecord类型对象 ， ProducerRecord 类提供了多种构造函数形参，常见的有如下三种 ：

ProducerRecord(topic,partition,key,value)；

ProducerRecord(topic,key,value)；

ProducerRecord(topic, value) ;

其中 topic 和 value 是必填的， partition 和 key 是可选的 。如果指定了 pa时tion，那么消息会被发送至指定的 partition ；如果没指定 partition 但指定了 Key，那么消息会按照 hash(key）发送至对应的 partition： 如果既没指定 partition 也没指定 key，那么 消息会按照 round-robin 模式发送（即以轮询的方式依次发送〉到每一个 partition。示例中将向 test-topic 主题发送三条消息。

3、消息消费者

和消息生产者类似，这里用 KafkaConsumer 类来创建一个消息消费者，该类的构造函数入参也是一系列属性值。

bootstrap servers 和生产者一样，表示 Kafka 集群。

groupid 表示消费者的分组 ID。

enableautocommit 表示 Consumer 的 offset 是否自 动提交 。

autocommitinterval ms 用于设置自动提交 offset 到 ZooKeeper 的时间间隔，时间单位是毫秒。

key deserializer 和 valuedeserializer 表示用字符串来反序列化消息数据。

消息消费者使用 subscribe 方法 订阅了 Topic 为 test-topic 的消息。 Consumer 调用poll 方法来轮询 Kafka 集群的消息， 一直等到 Kafka 集群中没有消息或达到超时时间（示例中设置超时时间为 100 毫秒）为止 。 如果读取到消息，则打印出消息记录的 pa此ition, offset、key 等。