Kafka学习笔记01【2024最新版】

一、kafka-课程介绍

B站视频学习地址：https://www.bilibili.com/video/BV1Gp421m7UN

官网地址：https://kafka.apache.org

kafka 3.7.0版本，作为经典分布式订阅、发布的消息传输中间件，kafka在实时数据处理、消息队列、流处理等领域具有广泛的应用场景。由于其高性能、高可靠、高吞吐，被数千家公司应用于高性能的数据管道、流分析和数据集成等不同场景，在海量实时数据传输和事件驱动的微服务架构中也被广泛地使用。

图形推演方式

学习步骤：

1、kafka集群启动

2、主题创建

3、生产消息

4、存储消息

5、消费消息

Java中，kafka与springboot进行集成

大数据场景中，我们将kafka和Flume、Spark和Flink等软件进行集成，那么kafka依然用作数据传输的消息中间件，那Flume就作为producer生产者来生产数据，而我们的Spark和Flink分布式计算引擎就作为consumer消费者来消费数据

三、kafka-软件介绍

kafka是一个由Scala与Java语言开发的分布式消息发布和订阅系统

从官网的描述上来看，kafka的核心特点就是高吞吐、高可靠、高可用、可伸缩，它的主要作用就是在大数据应用场景中用于分布式系统之间的数据交换

【需要搞明白一个问题】为什么分布式系统之间它需要使用一个软件来完成数据交换的这个过程？那说到我们数据交换啊，在Java开发的这个普通场景中，主要指的就是线程和线程之间的数据交换以及进程和进程之间的数据交换。

我们线程和线程之间是如何做数据交换的，其实我们主要是用内存来完成这个操作的。首先我们的Java虚拟机当中，每个线程呢其实是有它独立的这个栈内存空间的，每个线程是独立的，那么这两个我们的线程它们是如何交互的呢？首先它们两块内存是独立的，不过呢java虚拟机还有一块内存是它们共享的，这块内存呢我们称之为堆内存。咱们这个堆内存呢，其实就是我们所有线程共享的，那既然是所有线程共享的，那么是不是就可以将线程的数据发送到这块内存当中呢？这个肯定是没有问题的，那我们另外一个线程啊，其实就可以想办法呢，从这个堆内存当中把数据给我获取过来，因为你是共享的吗？你共享的话，你的数据放进去，我这边应该是可以获取到的，这个应该是没有问题的，那么为了数据的操作方便，Java还提供了专门的数据模型Q来作为数据的缓冲区进行数据的传输，就是我们通过一种队列的方式来完成，比较常见的呢就是阻塞队列啊BlockingQueue或者那个双端队列DQ啊等等。其实我们通过内存就可以实现我们数据的传输了，那么根本也不需要额外的第三方软件来完成。（缺点会导致内存不够用了，内存溢出）

接下来再说说进程和进程之间的数据交互，我们在执行Java程序的时候，就会启动一个java虚拟机的进程，那这个时候呢如果你想启用第二个进程的话，那它们两个能不能共享内存来完成我们的数据交互呢？其实是不行的，为什么呢？因为我们的java虚拟机啊，它在启动的时候会向操作系统申请内存，意味着两个不同的进程它们申请的内存其实是不一样的，那也就是说每个进程的内存空间是独立的，它们无法共享，因此我们进程之间传递数据呢都是采用网络数据流来进行传输的，比如我们使用Java提供的基础的socket、ServerSocket这样的接口来实现网络的数据传输（数据需要重复发送）

我们假设这个线程1往堆内存发数据，它的速度比较快，比如它的每秒钟会发我们的50条数据，我们接收数据的这个T2线程，它处理数据的这个速率是比较慢的，比如它每秒钟只能处理我们的30条数据，这样就会导致每秒钟其实是有20条数据是来不及处理的，它积压在我们的内存当中，那你可能会觉得这20条数据也不多呀，随着时间的推移，一旦积压的数据越来越多，内存就不够用了，就会产生内存溢出，从而导致服务不可用，那这个问题在进程之间的数据交互场景中，其实同样是存在的。那既然内存不够用，咱们将数据放到磁盘文件中不就好了吗，咱们这一块加上一个磁盘文件，反正也是作为数据的中转。磁盘确实比内存的存储数据要多得多，但是还是有上限的，所以内存、磁盘、网络我们一般都称为稀有资源，那么一旦数据的发送和接收的速率长时间的不均衡，就会因为数据积压导致资源被大量的消耗，进而对系统的吞吐量造成影响，导致系统服务不稳定，严重情况下，甚至会导致系统不可用

假设我们现在有三个进程，第一个进程用来生产数据，而我们的第二个和第三个进程，它用来接收数据，如果我们的第一个进程，它发送的数据需要同时发送给第二个进程和第三个进程，那么同一份数据它就得发两份，如果我们的数据的其中一部分，发送给第二个进程。那么另外一部分呢发送给我们第三个进程，那么我们的进程一就需要考虑对我的数据呢加一些我们的标记，来区分下游的每个不同进程，那么无形当中就需要增加一些逻辑来进行判断，来标记我们的数据，它的哪一部分发送给下游的哪个进程，而且无形中会增加我们第一个进程的逻辑处理难度，那么消耗更多的计算资源，从而降低系统的响应速度。而且你的数据可能是需要重复发送的，那么我们系统的吞吐量也会受到影响对吧。其实究其原因都是因为系统资源不够或进程之间它是直接进行数据交互所产生的，资源不够的问题呢？我们可以花钱买更好的资源来解决，但是最大的问题就是我们没有钱对吧。 咱们先考虑进程之间直接交互所产生的这些问题该如何解决，其实这里的进程的问题啊，其实就是耦合性的问题，那么耦合性的问题呢就是它们直接进行了交互吗？那我们只要不让它们直接交互也就是解耦合不就行了吗？可是我们如何让他们之间去解耦合呢？那这个时候我们就需要增加一个第三方的软件了，作为我们数据的一个缓冲区，你加上这个数据的缓冲区之后，不就降低了我们进程之间的耦合性了嘛。通过这个软件，我们可以对数据的产生和接收的速率进行调整，让整个数据的传输过程呢就更加的平稳高效了，那数据接收方呢也可以根据自己的业务场景呢，来动态从这个缓冲区中获取自己想要的数据，你是想全部都要呢，还是只要其中的一部分呢自己定就可以了，而我们的这个发送数据的这个进程啊，我只考虑往里面放就可以了，你看看逻辑不也变得简单了吗？中间的这个作为数据缓冲区的这个第三方软件，它设计的目的，就是为了进行数据的中转和临时存储，它并不会对数据进行处理，所以这个第三方软件也被叫做消息中间件，这就是分布式系统需要使用一个软件来完成数据交互的原因了

缓冲区【消息中间件】

四、kafka-JMS介绍

kafka软件它是一个消息中间件，那么作为数据传输的缓冲区，消息中间件它可以降低系统之间的耦合性，增强扩展性，那么在微服务的应用场景中，系统之间传递消息的时候呢，使用消息中间件的方式还是比较多的。在JavaEE的编程中为了规范这个消息中间件的开发和使用，那么Java提供了一套专门用于消息服务的应用程序接口

Java Message Service【JMS】，它其实跟我们访问数据库的时候，用到的JDBC的接口其实是非常类似的，我们在访问数据库的时候，我们可以通过JDBC规范中的接口来访问我们的数据。同样我们也可以通过JMS接口来进行数据的传输，接口的具体实现呢（不需要考虑）。在JMS规范中定义了很多用于通信的组件，首先我们先说一下这个message，在JMS中这个message其实就是消息的意思，在数据传输的过程中，我们传输的数据呢其实就是消息，消息中间件就可以把这个消息呢临时的给它存储起来。这个消息呢我们一般会采用Java的数据模型对它进行封装，会包含几个部分，首先第一个咱们叫消息头，还有呢我们是消息的一些相关属性，还有就是消息的主体内容，那么主体内容呢一般就是我们的数据。

中间的这个消息中间件在JMS中，它称之为叫消息的提供者【JMS Provider】，那么它指的就是提供消息服务的软件系统，这个系统的底层啊，一般会采用Java的队列Queue来实现数据的临时存储和数据传输。所以我们一般把这个提供消息服务的软件也称之为MQ软件，左边的process1它其实表述的就是发送数据的，所以在JMS啊，我们把它称之为叫生产者Producer，那么我们这边的这个进程呢用来获取数据，我们称之为消费者consumer。基本的消息传输的组件呢，它完整地体现了生产者和消费者模型，那所谓的生产者和消费者模型就是生产者和消费者它们彼此之间并不会直接通讯，而是通过中间的缓冲区进行中转来实现数据的传输和交互，那么就解决了生产者和我们消费者之间的强耦合问题。

因为kafka在内部都大量使用这种模型来解决组件之间的耦合性问题，说完这些组件的问题，我们再说一下JMS规范中对消息的发送和接收的一些要求，JMS支持两种消息发送和接收的模型，

P2P模型：点对点模型就是我们的消息队列中的每一条消息它只能被消费一次，也就是说无论多少个消费者来消费数据，那么我们一条数据，最终只能被一个消费者消费到。还有呢，就是当我们的producer来生产数据的时候，其实它是不用考虑我们的消费者是否存在的，它只要把我们的消息发送到provider当中就可以了，而我们的consumer呢在成功的接收到消息之后，它需要向发送者返回接收成功的消息，所以这边有个反馈的概念【feedback反馈】

PS模型：发布和订阅模型，事先我们需要将传输的这个数据给他进行一个分类，那这个分类我们称之为叫topic【主题】，那么也就意味着我们的producer生产数据的时候呢，它需要向指定的topic来发送数据。讲道理，对某一个类型数据感兴趣的消费者consumer，它就可以订阅相关的主题了，那么从订阅的主题中获取数据，如果消费者没有订阅相关的主题的话，那么它是无法消费其中的数据的，consumer是可以有多个的，那既然有多个的话，它们就可以同时订阅这个主题，那么这个主题的数据就可以被他们同时消费掉

RabbitMQ、ActiveMQ、RocketMQ、Kafka，它们基本上都遵循了JMS相关的协议，虽然kafka拥有作为一个消息系统的应该具备的功能，但是却有着自己独特的设计。可以这样说，kafka借鉴了JMS规范的思想，但是并没有完全遵循JMS规范

五、kafka-组件

kafka没有完全遵循JMS规范，但还是借鉴了JMS规范的思想，对它的通信组件呢和我们之前讲的JMS基本上还是一致的。

message，在JMS中我们传递的数据其实就是传递的消息，只不过呢？在kafka中就不叫message了，然们叫record【记录】

中间的JMS Provider就是kafka了，作为消息中间件，kafka通过启动一个独立的进程来提供消息的临时存储用于传输数据，所以我们一般会将这个kafka的进程称之为叫消息服务节点，不过因为本质上呢这个节点呢，它只是用于中转数据和临时存储数据，它并不会对数据进行什么处理啊，所以我们把这个称之为叫代理或者叫中介，所以它们叫kafka的一个broker，这个broker就是中介代理的意思

咱们这里只是一套规范接口，所以我们的程序呢需要调用这个接口，通过调用咱们kafka的producer的这个接口来实现我们数据的生产

Topic主题，因为主题本质就是对数据的分类，所以我们这里可以简单的理解为就是kafka节点的临时存储数据的一个缓冲区，它的底层一般都是用队列吗？队列其实就是内存当中的一个数据模型。所以很明显，内存是无法长期的存储数据的，一旦咱们的服务节点出现了故障，重启的话，那么咱们的内存就会重新分配数据就会容易丢失，所以为了数据的可靠性更高一点，我们这里仅仅用我们内存来作为缓冲区实现数据的传输其实是不够的。那么好的方式应该就是将数据同时的保存到磁盘文件中，那么这样的话既保证了数据传输的高效，又保证了数据的安全可靠。

kafka就会将数据保存到咱们的文件当中

消息的中间件它不仅仅用作数据传输的载体，还要保证数据的顺序不被打乱，所以kafka会给临时存储的数据增加标记来方便数据的有序访问，这个标记它有点类似于咱们java中数组的索引，从零开始给它一个0、1、2，这样的话我们的每个数据增加了一个标记，那么它的顺序就定了，那么我们在消费数据的时候呢，就可以指定某一个顺序、某一个标记来进行访问了，我们称之为偏移量

六、kafka-安装与启动

windows安装步骤：

1、解压 kafka_2.12-3.7.0.tgz

2、kafka_2.12-3.7.0目录下新建data目录

3、修改config目录下的zookeeper.properties

dataDir=D:/xxx/kafka_2.12-3.7.0/data/zk

4、在kafka_2.12-3.7.0目录下新建zk.cmd文件

call bin/windows/zookeeper-server-start.bat config/zookeeper.properties

5、双击zk.cmd，启动zookeeper

6、修改config目录下的server.properties

log.dirs=D:/gitee/kafka_2.12-3.7.0/data/kafka

7、在kafka_2.12-3.7.0目录下新建kfk.cmd文件

call bin/windows/kafka-server-start.bat config/server.properties

8、双击kfk.cmd，启动kafka

十三、kafka-基础架构图形推演

数据只保存在内存中是不安全的，我们需要把数据保存到日志文件中。为什么是日志文件呢？就是因为kafka早期就是做日志处理传输的，所以它的文件就叫做.log

架构

kafka采用生产者消费者的模型，所以它是允许多个生产者来生产数据，同时呢它也允许多个消费者来消费数据

有没有发现什么问题？我们当前的这个Kafka Broker 有很多的生产者和消费者对它进行访问，这个频繁的数据请求访问呢？大概率就会因为吞吐量过大产生IO热点问题，从而导致这个单一的这个节点成为整个分布式系统的性能瓶颈，进行就会降低系统的可用性和稳定性，一旦我们当前的Broker它宕掉了，那我们的数据其实就无法访问了，你连服务都没有了，怎么可能访问数据呢？甚至可能还会因为我们的内存中的数据还没有来得及写入咱们的磁盘文件而导致数据丢失，这明显它不是一个好的系统架构方案

横向扩展：增加服务节点，搭建服务器集群，降低单点故障带来的风险（增加多个Broker节点，把Topic划分为多个数据块【添加编号进行区分】，放在多个Broker节点，消费者添加消费者组）

我们增加了多个broker节点的话，是不是就能够解决问题，我们的多个节点它其实同时提供咱们的数据服务的话，那么你的请求不久可以均匀地分布到不同的节点上面去，通过这样的方式就可以在一定程度上缓解咱们所谓的IO热点问题（一定程度上缓解，但是并没有说真正的解决），因为我们的生产者是需要向指定的主题发送数据的，而我们的消费者需要从指定的主题来获取数据，咱们的这个主题的数据啊，它就存在这个地方来看一下，那么我们的生产者那还是要指向这一个位置呀，你即使有三个broker，那不还是要指向其中的一个吗？同样道理，那么你的主题在这里，那么我的消费者就想消费这个主题的数据，那么大家有没有发现你这么做，其实即使你增加了多个节点，依然没有解决我们的IO的热点问题。

我们就应该把Topic的这个数据的分类给它切成不同的数据块，来放置在不同的broker节点，这样的话大家会发现你要想去生产数据，你要想去消费数据的话，可就不一定只指向这一个broker的对吗？我们要把我们的一个Topic给它分成几块，那么我的这个队列它应该也是存在的

那这个时候呢我们再去访问数据的时候，我们生产数据就可以指向不同的broker，但是是同一个主题不就完事了吗？通过这种方式就可以有效地解决咱们之前的那个IO热点问题

我们当前把一个Topic给它切成几块，那么这几块其实是个整体，那既然是个整体，可是我们的数据呢却把它划分开了，那么我们怎么来区分着不同的数据块呢？给不同的数据块增加一个编号，这个编号从零开始，我们这里的数据块呢我们就称之为叫分区partition。我们的不同数据块呢，需要增加一个标记来做区分，咱们这个标记就是个顺序号。

你有三个分区了，我们的kafka的producer，其实就可以把数据发向不同的分区，同一个主题，但是分区它不一样，我就可以发送给不同的分区。我们的消费者，它就可以消费同一个主题不同分区的数据，那你的这个请求就可以分布到不同的节点，消费的数据并不是完整的啊，我去订阅一个主题，那我怎么可能只去消费一部分数据，那就意味着我们的消费者他得同时能消费0、1、2分区，那你这不等于发送了更多的请求吗？那你这个的性能其实还是有问题的。

我们的kafka他提出了一个概念，这个概念就是一个组的概念，就是把我们的多个消费者当成一个整体来对我们的主题消费

消费者组其实专门就是用来消费咱们的数据的，他们是一个整体，你消费他他消费他他消费他，你们三个消费者消费的数据就是完整的Topic的数据，前面讲过的代码就有group这个配置，其实就是我们消费者组的那个id，给大家说了数据的分布问题，接下来咱们来说说数据的存储问题。

前面也提到了数据的传输不仅仅需要将数据放在缓存当中，提高我们的数据传输效率，还需要将数据保存到磁盘文件当中，保证数据的可靠性，不过这个可靠性是相对的，我们当前的这个log，其实我们每一个broker都应该有一个log

我们假设某一个我们的broker节点，它down掉了，你的这个日志文件中数据你还能消费到吗？你这个服务器都down掉了，那么只剩下我们的上面的这个0、2号分区，你没有的情况下，你的数据不就不完整了吗？

我们确实还是采用备份的方式，但是我们这个备份就不要放在自己的这个broker上面去了，你放在别的地方不就可以了吗？把备份的日志文件数据放在其它的broker节点即可。假设咱们当前的某个节点咱们down掉了，其实这个节点如果能够运行的话，那么我们的数据并没有丢失，那这样的话可靠性就会更强一些

纵向扩展：增加系统的资源配置，比如使用IO效率更好的固态硬盘以及更大的内存、更多的CPU核，还有更快的网络（其实无法从根本上解决问题的，咱们的资源是有限的，一旦单机的吞吐量超过阈值，无论你增加什么样的资源都是无法解决问题的）

总结：

为了数据可靠性，可以将数据文件进行备份，但是kafka没有备份的概念

kafka中称之为副本，多个副本同时只能有一个提供数据的读写操作，其他副本只是用于备份

具有读写能力的副本称之为Leader副本，作为备份的副本称之为Follower副本

Broker： 服务节点（集群）

Partition: 分区（编号）

副本：Leader和Follower

十四、Kafka-基础组件图形推演

为了能够很好地监控集群中各个节点的运行情况，提高集群服务的可用性，我们一般会在集群中找一个节点，作为整个集群的管理者

每一个broker都能做备份【Standby】，就是当前集群的任何节点都可以做管理者，你不是down掉了吗？他们几个都有可能去做管理者，就是任何一个节点都可以做备份

选举：使用zookeeper，其中有一个功能就是用来做选举功能的

十六、kafka-启动服务-zookeeper的核心功能

启动服务器，创建主题，然后生产数据，消费数据的这个顺序

第一个就是zookeeper的ZNode节点功能，kafka是客户端，zookeeper就是服务器，kafka作为客户端去连接咱们当前的zookeeper，创建它的ZNode节点，这个ZNode节点，它有点类似于咱们文件系统中的那个路径的那个概念

kafka这个客户端，它连接到zookeeper以后，就创建了我们当前不同的节点。咱们创建好之后，kafka就可以对我们的每一个ZNode节点进行一些数据的存储和访问，不过呢这个节点它只能创建一次。我们的kafka创建某一个节点只会创建一次，一旦创建成功了，另一个kafka想创建就不太可能了。

这个ZNode节点其实还会有分类，一般分为持久化节点和临时节点，所谓的持久化节点是什么意思？意味着咱们这个kafka作为客户端连接到zookeeper之后，我们创建这个节点，如果这个时候我们的kafka切断了咱们的连接，关闭了咱们的连接，这个节点会一直存在，这个就叫持久化节点

什么叫临时节点？当我们的kafka切断了跟zookeeper之间的连接之后，那这个节点就会自动被删除，这个就叫做临时节点

第二个功能就是ZNode节点有watch的监听功能，咱们这个kafka连接上zookeeper之后，其实是可以对某一个节点进行监听的，可以监听到节点的变化，咱们数据的变化或者连接超时，这些我们都可以监听到。一旦监听到，kafka往这个节点增加一些数据或者修改了一些数据甚至可能会删除一些我们的数据，那这个时候我们的他【ZNode节点】就会被监听到，那么监听到以后，马上就会通知我们的这个kafka，让他去做一些相应的处理

callback【回调】，所谓的回调就是当我们这边监听到一些事件的变化的时候，我们就会让我们的监听的kafka的客户端做出一些相应的处理，所以这就是我们的zookeeper所提供的一些功能，而kafka恰恰就利用这些功能来做集群内部的管理

十七、kafka-启动服务-Broker启动后Zookeeper节点的变化

安装prettyZoo-win.msi

十八、kafka-启动服务-Controller的选举

Controller选举的简单流程：首先我们的这个zookeeper当中没有任何的节点，我们先启动第一个broker的id为1的一个节点，这个时候我们就会在主题当中去创建一个临时节点，咱们叫id=1，就意味着绿色节点当成了我们整个集群的管理者。那么这个时候我们的第二个节点开始启动，它启动以后，因为已经有那个管理者了，所以它就没办法了，怎么办呢？它这个时候就会在我们这儿创建一个监听器，监听器的目的就很简单，监听节点的变化。同样的道理，Broker id=3它也会去创建监听器来监听我们的变化。那这个时候一旦我们的broker id=1，它挂掉了，失去连接，咱们当前的这个节点就会被删除。这个时候Broker id2和3就会发出新的请求，就会跟zookeeper创建连接，谁先建立连接，谁先创建节点，那么谁就是整个集群的管理者，所以我们说假设红色它创建成功

十九、kafka-启动服务-Controller和Broker的通信原理及底层通信机制