Kafka架构高性能和高可用性分析

Posted 2022-02-25 aijiudu

目录

一、什么是Kafka？

二、Kafka的使用场景

三、kafka系统架构

四、Kafka高性能

1 批量发送消息

2 持久化消息-顺序写

3 基于索引文件的查询

4 零拷贝

五、Kafka高可靠

1 消息备份

2 ISR & LEO & HW

3 Acks

六、MAC本地安装Kafka

1 安装

2 启动kafka服务

3 创建Topic，显示数据 

本文主要介绍Kafka架构、高性能、高可用以及mac本地安装kafka，本文参考网上材料和kafka书籍，学习总结，分享出来希望能帮到大家，如有问题及时指出。

一、什么是Kafka？

Kafka是一个消息队列，把消息放到队列里边的叫生产者，从队列里边消费的叫消费者。Kafka虽然是基于磁盘做的数据存储，但却具有高性能、高吞吐、低延时、可持久化、可水平扩展、支持流数据处理等特点。

消息队列（Message Queue）是一种进程间通信或同一进程的不同线程间的通信方式，主要解决应用耦合、异步消息、流量削锋等问题。实现高性能、高可用、可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。

Kafka的特性

Kafka是一种分布式的，基于发布/订阅的消息系统，主要特性如下：

特性	分布式	高性能	持久性和扩展性
描述	多分区	高吞吐量	数据可持久化
	多副本	低延迟	容错性
	多订阅者	高并发	支持水平在线扩展
	基于ZooKeeper调度	时间复杂度为O(1)	消息自动平衡

二、Kafka的使用场景

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

（2）消息系统：解耦、冗余存储、流量削峰、缓存数据、异步处理等。

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

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

（5）流式处理：完整的流处理库，窗口连接、变换、聚合和spark streaming、storm、flink配合使用。

消息队列（系统）的基本使用场景：

三、kafka系统架构

Kafka系统架构

Producer生产者/Consumer消费者：Kafka是一个消息队列，把消息放到队列里边的叫生产者，从队列里边消费的叫消费者。

topic：一个消息中间件，队列不单单只有一个，我们往往会有多个队列，而我们生产者和消费者就得知道：把数据丢给哪个队列，从哪个队列消息。我们需要给队列取名字，叫做topic(相当于数据库里边表的概念)。给队列取了名字以后，生产者就知道往哪个队列丢数据了，消费者也知道往哪个队列拿数据了。我们可以有多个生产者往同一个队列(topic)丢数据，多个消费者往同一个队列(topic)拿数据

Partition：为了提高一个队列(topic)的吞吐量，Kafka会把topic进行分区(Partition)。

Topic & Partition

Topic在逻辑上可以被认为是一个Queue，Kafka中每条消息都必须指定一个Topic，一个Topic中的消息可以分布在集群中的多个Broker中，Consumer根据订阅的Topic到对应的Broker上去拉取消息。为了提升整个集群的吞吐量，物理上一个Topic可以分成多个Partition，每个Partition在磁盘上对应一个文件夹，该文件夹下存放了这个Partition的所有消息文件和索引文件。

 Broker：一台Kafka服务器叫做Broker，Kafka集群就是多台Kafka服务器，一个topic会分为多个partition，实际上partition会分布在不同的broker中。

总结：生成者往topic丢数据，实际上数据会再partition中，partition分布在不同的Broker（服务器）上。

备份：分布式肯定会带来问题：“万一其中一台broker(Kafka服务器)出现网络抖动或者挂了，怎么办？”数据存在不同的partition上，那kafka就把这些partition做备份。备份散落在不同的broker上。备份分区仅仅用作于备份，不做读写。如果某个Broker挂了，那就会选举出其他Broker的partition来作为主分区，这就实现了高可用。

Consumer Group：消费者实际上也是从partition中取，下面图的情况，是一个消费者消费两个分区的数据。多个消费者可以组成一个消费者组。消费组1有三个消费者，因为只有两个partition，所以一个空闲，消费组2只有一个消费者，就要消费两个partition。无论是新增的消费者组还是原本的消费者组，都能消费topic的全部数据。

offset：Kafka就是用offset来表示消费者的消费进度到哪了，每个消费者会都有自己的offset。说白了offset就是表示消费者的消费进度。

 这里要注意，因为Kafka读取消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高Kafka性能无关。另外，Kafka会为每一个Consumer Group保留一些metadata信息（当前消费的消息的位置，即offset）。这个offset由Consumer控制，Consumer会在消费完一条消息后递增该offset。当然，Consumer也可将offset设成一个较小的值，重新消费一些消息。因为offet由Consumer控制，所以Kafka Broker是无状态的，它不需要标记消息是否被消费过，也不需要通过Broker去保证同一个Consumer Group只有一个Consumer能消费某一条消息，因此也就不需要锁机制，从而保证了Kafka的高吞吐率。

四、Kafka高性能

Producer生产消息，以Partition的维度，按照一定的路由策略，提交消息到Broker集群中各Partition的Leader节点，Consumer以Partition的维度，从Broker中的Leader节点拉取消息并消费消息。

Producer生产消息会涉及大量的消息网络传输，如果Producer每生产一个消息就发送到Broker会造成大量的网络消耗，严重影响到Kafka的性能。为了解决这个问题，Kafka使用了批量发送的方式。 Broker在持久化消息、读取消息的时候，如果采用传统的IO读写方式，会严重影响Kafka的性能，为了解决这个问题，Kafka采用了顺序写+零拷贝的方式。

下面分别从批量发送消息、持久化消息、零拷贝三个角度介绍Kafka如何提高性能。

1 批量发送消息

 Kafka通过将Topic划分成多个Partition，如下图所示，消息经过路由策略，被分发到不同的Partition对应的本地队列（序列化消息并压缩消息后，追加到本地的记录收集器(RecordAccumulator），Sender不断轮询记录收集器，当满足一定条件时，将队列中的数据发送到Partition Leader节点。

路由策略：Kafka的路由策略主要有三种：

• Round Robin：Producer将消息均衡地分配到各Partition本地队列上，是最常用的分区策略。

• 散列：Kafka对消息的key进行散列，根据散列值将消息路由到特定的Parttion上，键相同的消息总是被路由到相同的Partition上。

• 自定义分区策略：Kafka支持自定义分区策略，可以将某一系列的消息映射到相同的Partition。

本地队列：

Producer会为每个Partition都创建一个双端队列来缓存客户端消息，队列的每个元素是一个批记录(ProducerBatch)，批记录使用createdMs表示批记录的创建时间(批记录中第一条消息加入的时间)， topicPartion表示对应的Partition元数据。当Producer生产的消息经过序列化，会被先写入到recordsBuilder对象中。满足发送条件，就会被Sender发送到Partition对应的Leader节点。

Sender发送数据到Broker的条件有两个：

• 消息大小达到阈值

• 消息等待发送的时间达到阈值

Sender：Sender读取记录收集器，得到每个Leader节点对应的批记录列表，找出准备好的Broker节点并建立连接，然后将各个Partition的批记录发送到Leader节点。

//Sender读取记录收集器，按照节点分组，创建客户端请求，发送请求
public void run(long now) 
	Cluster cluster = metadata.fetch();
  //获取准备发送的所有分区
  ReadCheckResult result = accumulator.ready(cluster, now);
  //建立到Leader节点的网络连接，移除还没有准备好的节点
  Iterator<Node> iter = result.readyNodes.iterator();
  while(iter.hasNext()) 
  	Node node = iter.next();
    if (!this.client.read(node, now)) 
    	iter.remove();
    
    //读取记录收集器，返回的每个Leader节点对应的批记录列表，每个批记录对应一个分区
    Map<Integer, List<RecordBatch>> batches = accumulator.drain(cluster, result.readyNodes, 
    																													  this.maxRequestSize, now);
    //以节点为级别的生产请求列表，即每个节点只有一个客户端请求
    List<ClientRequest> requests = createProduceRequests(batches, now);
    for (ClientRequest request : requests) 
    	client.send(request, now);
    
    //这里才会执行真正的网络读写，比如将上面的客户端请求发送出去
    this.client.poll(pollTimeout, now);
  

具体的步骤如下：

1、消息被记录收集器收集，并按照Partition追加到队列尾部一个批记录中。

2、Sender通过ready()从记录收集器中找出已经准备好的服务端节点，规则是Partition等待发送的消息大小和等待发送的时间达到阈值。

3、节点已经准备好，如果客户端还没有和它们建立连接，通过connect()建立到服务端的连接。

4、Sender通过drain()从记录收集器获取按照节点整理好的每个Partition的批记录。

5、Sender得到每个节点的批记录后，为每个节点创建客户端请求，并将消息发送到服务端。

消息格式：每个消息文件都是一个log entry序列，其格式如下图所示：

上图左侧的“RECORD”部分就是Kafka的消息格式，一条完整的消息包含RECORD、offset以及message size。其中offset用来标识它在Partition中的偏移量，这个offset是逻辑值，而非实际物理偏移值，message size表示消息的大小。与消息对应的还有消息集的概念，消息集中包含一条或者多条消息，消息集不仅是存储于磁盘以及在网络上传输（Produce & Fetch）的基本形式，而且是kafka中压缩的基本单元，详细结构参考上图右侧。下面来具体描述一下消息（RECORD）格式中的各个字段，从crc32开始算起，各个字段的解释如下：

crc32（4B）：crc32校验值，校验范围为magic至value之间。

magic（1B）：消息格式版本号，0.9.X版本的magic值为0。

attributes（1B）：消息的属性，总共占1个字节，低3位表示压缩类型：0表示NONE、1表示GZIP、2表示SNAPPY、3表示LZ4，其余位保留。

key length（4B）：表示消息的key的长度。如果为-1，则表示没有设置key，即key=null。

key：可选，如果没有key则无此字段。

value length（4B）：实际消息体的长度。如果为-1，则表示消息为空。

value：消息体，可以为空。

2 持久化消息-顺序写

Broker中需要将大量的消息做持久化，而且存在大量的消息查询场景，如果采用传统的IO操作，会带来大量的磁盘寻址，影响消息的查询速度，限制了Kafka的性能。为了解决这个问题，Kafka采用顺序写的方式来做消息持久化。

Topic是一个逻辑概念，如下图：partition命名：opic为topic_test有三个partition，partition的命名是topic名字+序号，是个文件夹，在log.dirs配置的目录下。partition可以细分为Segment，Segment文件由两部分组成
分别为index文件和log文件。

 写消息：Producer传递到Broker的消息集中的每条消息都会分配一个顺序值（只是相对于本批次的序号）用来标记Producer所生产消息的顺序，每一批消息的顺序值都从0开始。服务端会将每条消息的顺序值转换成绝对偏移量(Broker从Partition维度来标记消息的顺序，用于控制Consumer消费消息的顺序)。Kafka通过nextOffset(下一个偏移量)来记录存储在日志中最近一条消息的偏移量。消息发送到Broker后，每条消息都被顺序写该Partition所对应的文件中，因此效率非常高，这是Kafka高吞吐率的一个很重要的保证。

下图给出一个例子，Producer写到Partition的消息有3条消息，对应的顺序值是[0,1,2]。消息写入前，nextOffset是899，msg0、msg1、msg2是连续写入的三条消息，消息被写入后其绝对偏移量分别是900、901、902，对应的顺序值分别是0、1、2，nextOffset变成902。

3 基于索引文件的查询

Kafka通过索引文件提高对磁盘上消息的查询效率。

如上图所示：假设有1000条消息，每100条消息写满了一个日志分段，一共会有10个日志分段。客户端要查询偏移量为938的消息内容，如果没有索引文件，我们必须从第一个日志分段的数据文件中，从第一条消息一直往前读，直到找到偏移量为999的消息。有了索引文件后，我们可以在最后一个日志分段的索引文件中，首先使用绝对偏移量999减去基准偏移量900得到相对偏移量99，然后找到最接近相对偏移量99的索引数据90，相对偏移量90对应的物理地址是1365，然后再到数据文件中，从文件物理位置1365开始往后读消息，直到找到偏移量为999的消息。

Kafka的索引文件的特性：

• 索引文件映射偏移量到文件的物理位置，它不会对每条消息都建立索引，所以是稀疏的。

• 索引条目的偏移量存储的是相对于“基准偏移量”的“相对偏移量” ，不是消息的“绝对偏移量” 。

• 偏移量是有序的，查询指定的偏移量时，使用二分查找可以快速确定偏移量的位置。

• 指定偏移量如果在索引文件中不存在，可以找到小于等于指定偏移量的最大偏移量。

• 稀疏索引可以通过内存映射方式，将整个索引文件都放入内存，加快偏移量的查询。

由于Broker是将消息持久化到当前日志的最后一个分段中，写入文件的方式是追加写，采用了对磁盘文件的顺序写。对磁盘的顺序写以及索引文件加快了Broker查询消息的速度。

4 零拷贝

Kafka中存在大量的网络数据持久化到磁盘(Producer到Broker)和磁盘文件通过网络发送(Broker到Consumer)的过程。这一过程的性能直接影响到Kafka的整体性能。Kafka采用零拷贝这一通用技术解决该问题。

零拷贝技术可以减少数据拷贝和共享总线操作的次数，消除传输数据在存储器之间不必要的中间拷贝次数，减少用户应用程序地址空间和操作系统内核地址空间之间因为上下文切换而带来的开销，从而有效地提高数据传输效率。

（1）传统的文件拷贝

传统文件拷贝：数据的四次拷贝与四次上下文切换

4次拷贝：

1. 一个读操作发生后，CPU/DMA执行了一次数据拷贝，数据从磁盘拷贝到内核空间；

2. cpu将数据从内核空间拷贝至用户空间

3. 调用send()，cpu发生第三次数据拷贝，由cpu将数据从用户空间拷贝至内核空间(socket缓冲区)

4. send()执行结束后，CPU/DMA执行第四次数据拷贝，将数据从内核拷贝至协议引擎

4 次上下文切换：

1. read 系统调用时：用户态切换到内核态；

2. read 系统调用完毕：内核态切换回用户态；

3. write 系统调用时：用户态切换到内核态；

4. write 系统调用完毕：内核态切换回用户态；

DMA 技术：DMA【协处理器（一块独立的芯片）】负责内存与其他组件之间的数据拷贝，CPU 仅需负责管理，而无需负责全程的数据拷贝；DMA仅仅能用于设备间交换数据时进行数据拷贝，但是设备内部的数据拷贝还需要 CPU 来亲力亲为。

Kafka 作为一个消息队列，涉及到磁盘 I/O 主要有两个操作：

• Provider 向 Kakfa 发送消息，Kakfa 负责将消息以日志的方式持久化落盘；(mmap 机制)

• Consumer 向 Kakfa 进行拉取消息，Kafka 负责从磁盘中读取一批日志消息，然后再通过网卡发送；

（2）kafka provider持久化数据（写入）采用memory map-零拷贝技术（mmap 也至少需要 4 次上下文切换）

零拷贝技术是一个思想，指的是指计算机执行操作时，CPU 不需要先将数据从某处内存复制到另一个特定区域。零拷贝不是不进行拷贝，而是 CPU 不再全程负责数据拷贝时的搬运工作。

mmap：仅代替 read 系统调用，将内核空间地址映射为用户空间地址，write 操作直接作用于内核空间。通过 DMA 技术以及地址映射技术，用户空间与内核空间无须数据拷贝，实现了 zero copy

mmap 技术有如下特点：

1. 利用 DMA 技术来取代 CPU 来在内存与其他组件之间的数据拷贝，例如从磁盘到内存，从内存到网卡；

2. 用户空间的 mmap file 使用虚拟内存，实际上并不占据物理内存，只有在内核空间的 kernel buffer cache 才占据实际的物理内存；

3. mmap() 函数需要配合 write() 系统调动进行配合操作，这与 sendfile() 函数有所不同，后者一次性代替了 read() 以及 write()；因此 mmap 也至少需要 4 次上下文切换；

4. mmap 仅仅能够避免内核空间到用户空间的全程 CPU 负责的数据拷贝，但是内核空间内部还是需要全程 CPU 负责的数据拷贝；

利用 mmap() 替换 read()，配合 write() 调用的整个流程如下：

1. 用户进程调用 mmap()，从用户态陷入内核态，将内核缓冲区映射到用户缓存区；

2. DMA 控制器将数据从硬盘拷贝到内核缓冲区（可见其使用了 Page Cache 机制）；

3. mmap() 返回，上下文从内核态切换回用户态；

4. 用户进程调用 write()，尝试把文件数据写到内核里的套接字缓冲区，再次陷入内核态；

5. CPU 将内核缓冲区中的数据拷贝到的套接字缓冲区；

6. DMA 控制器将数据从套接字缓冲区拷贝到网卡完成数据传输；

7. write() 返回，上下文从内核态切换回用户态。

（3）kafka Consumer拉取数据（读）采用sendfile-零拷贝技术

sendfile：一次代替 read/write 系统调用，通过使用 DMA 技术以及传递文件描述符代替数据拷贝，实现了 zero copy。

sendfile流程如下：

1. 将文件拷贝到page cache中；

2. 向socket buffer中追加当前要发生的数据在page cache中的位置和偏移量；

3. 根据socket buffer中的位置和偏移量直接将page cache的数据copy到网卡设备中；

kafka读数据采用sendfile机制优点：

• sendfile 依赖于 DMA 技术，将四次CPU全程负责的拷贝与四次上下文切换减少到两次，避免了内核空间到用户空间的CPU全程负责的数据移动；

• sendfile 基于 Page Cache 实现，因此如果有多个 Consumer 在同时消费一个主题的消息，那么由于消息一直在 page cache 中进行了缓存，因此只需一次磁盘 I/O，就可以服务于多个 Consumer；

缺点：因为用户线程根本就不能够通过 sendfile 系统调用得到传输的数据（传递文件描述符代替数据拷贝：page cache 以及 socket buffer 都在内核空间中；数据在传输中没有被更新；），所以无法修改数据。

五、Kafka高可靠

1 消息备份

Kafka允许同一个Partition存在多个消息副本(Replica)，每个Partition的副本通常由1个Leader及0个以上的Follower组成，生产者将消息直接发往对应Partition的Leader，Follower会周期地向Leader发送同步请求，Kafka的Leader机制在保障数据一致性地同时降低了消息备份的复杂度。

同一Partition的Replica不应存储在同一个Broker上，因为一旦该Broker宕机，对应Partition的所有Replica都无法工作，这就达不到高可用的效果。为了做好负载均衡并提高容错能力，Kafka会尽量将所有的Partition以及各Partition的副本均匀地分配到整个集群上。

2 ISR & LEO & HW

ISR(In-Sync Replicas)指的是一个Partition中与Leader“保持同步”的Replica列表(实际存储的是副本所在Broker的BrokerId)，这里的保持同步不是指与Leader数据保持完全一致，只需在replica.lag.time.max.ms时间内与Leader保持有效连接，官方解释如下

If a follower hasn't sent any fetch requests or hasn't consumed up to the leaders log end offset for at least this time, the leader will remove the follower from isr,( default value =10000 )

Follower周期性地向Leader发送FetchRequest请求(数据结构见下)，发送时间间隔配置在replica.fetch.wait.max.ms中，默认值为500。

public class FetchRequest 
    private final short versionId;
    private final int correlationId;
    private final String clientId;
    private final int replicaId;
    private final int maxWait;    // Follower容忍的最大等待时间: 到点Leader立即返回结果，默认值500
    private final int minBytes;   // Follower容忍的最小返回数据大小：当Leader有足够数据时立即返回，兜底等待maxWait返回，默认值1
    private final Map<TopicAndPartition, PartitionFetchInfo> requestInfo;  // Follower中各Partititon对应的LEO及获取数量

 各Partition的Leader负责维护ISR列表并将ISR的变更同步至ZooKeeper，被移出ISR的Follower会继续向Leader发FetchRequest请求，试图再次跟上Leader重新进入ISR。

ISR中所有副本都跟上了Leader，通常只有ISR里的成员才可能被选为Leader。当Kafka中unclean.leader.election.enable配置为true(默认值为false)且ISR中所有副本均宕机的情况下，才允许ISR外的副本被选为Leader，此时会丢失部分已应答的数据。

每个Kafka副本对象都有下面两个重要属性：

LEO(log end offset) ，即日志末端偏移，指向了副本日志中下一条消息的位移值(即下一条消息的写入位置)

HW(high watermark)，即已同步消息标识，因其类似于木桶效应中短板决定水位高度，故取名高水位线

所有高水位线以下消息都是已备份过的，消费者仅可消费各分区Leader高水位线以下的消息，对于任何一个副本对象而言其HW值不会大于LEO值

Leader的HW值由ISR中的所有备份的LEO最小值决定(Follower在发送FetchRequest时会在PartitionFetchInfo中会携带Follower的LEO)

下图详细的说明了当Producer生产消息至Broker后，ISR以及HW和LEO的流转过程： 

3 Acks

为了讲清楚ISR的作用，下面介绍一下生产者可以选择的消息应答方式，生产者发送消息中包含acks字段，该字段代表Leader应答生产者前Leader收到的应答数

• acks=0

生产者无需等待服务端的任何确认，消息被添加到生产者套接字缓冲区后就视为已发送，因此acks=0不能保证服务端已收到消息，使用场景较少，本文不做任何讨论

• acks=1

Leader将消息写入本地日志后无需等待Follower的消息确认就做出应答。如果Leader在应答消息后立即宕机且其他Follower均未完成消息的复制，则该条消息将丢失 

上图左侧的稳态场景下，Partition1的数据冗余备份在Broker0和Broker2上；Broker0中的副本与Leader副本因网络开销等因素存在1秒钟同步时间差，Broker0中的副本落后124条消息；Broker2中的副本存在8秒钟同步时间差，Broker2中的副本落后7224条消息。若图中的Broker1突然宕机且Broker0被选为Partition1的Leader，则在Leader宕机前写入的124条消息未同步至Broker0中的副本，这次宕机会造成少量消息丢失。

• acks=all

Leader将等待ISR中的所有副本确认后再做出应答，因此只要ISR中任何一个副本还存活着，这条应答过的消息就不会丢失。acks=all是可用性最高的选择，但等待Follower应答引入了额外的响应时间。Leader需要等待ISR中所有副本做出应答，此时响应时间取决于ISR中最慢的那台机器，下图中因复制产生的额外延迟为3秒。

 Broker的配置项min.insync.replicas(默认值为1)代表了正常写入生产者数据所需要的最少ISR个数，当ISR中的副本数量小于min.insync.replicas时，Leader停止写入生产者生产的消息，并向生产者抛出NotEnoughReplicas异常，阻塞等待更多的Follower赶上并重新进入ISR。被Leader应答的消息都至少有min.insync.replicas个副本，因此能够容忍min.insync.replicas-1个副本同时宕机。

小结：发送的acks=1消息会出现丢失情况，为不丢失消息可配置生产者acks=all & min.insync.replicas >= 2

kafka故障恢复，Broker故障场景分析，Controller故障恢复本文不做介绍。

六、MAC本地安装Kafka

本地安装kafka创建topic、producer和consumer

1 安装

直接使用brew安装kafka

brew install kafka

2 启动kafka服务

我们需要先后启动zookeeper和kafka服务。

它们都需要进入 /usr/local/Cellar/kafka/3.1.0目录，然后再启动相应的命令。
cd /usr/local/Cellar/kafka/3.1.0
启动zookeeper服务，运行命令：
zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties
启动kafka服务，运行命令：
kafka-server-start /usr/local/etc/kafka/server.properties

3 创建Topic，显示数据 

新版本的kafka，已经不需要依赖zookeeper来创建topic，新版的kafka创建topic指令为下：
/usr/local/Cellar/kafka/3.1.0/bin$ kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_demo
查看topic
kafka-topics --list --bootstrap-server localhost:9092
生产者：
wangzhibin:/usr/local/Cellar/kafka/3.1.0/bin$ kafka-console-producer --broker-list localhost:9092 --topic topic_demo
>hahha
>kafka
>spark
查看topic文件：cd /usr/local/var/lib/kafka-logs/topic_demo-0
消费者：
wangzhibin:/usr/local/Cellar/kafka/3.1.0/bin$ kafka-console-consumer --bootstrap-server localhost:9092 --topic topic_demo --from-beginning
hahha
kafka
spark
暂停kafka
kafka-server-stop