第一章 kafka入门笔记

电玩女神 2023-06-08 15:59 452阅读 0赞

# **一、Kafka简介** #

## **1、什么是Kafka** ##

kafka最初是LinkedIn的一个内部基础设施系统。最初开发的起因是LinkedIn虽然有了数据库和其他系统可以用来存储数据，但是缺乏一个可以帮助处理持续数据流的组件。所以在设计理念上，开发者不想只是开发一个能够存储数据的系统，如关系数据库Nosql数据库、搜索引擎等等，更希望把数据看成一个持续变化和不断增长的流，并基于这样的想法构建出一个数据系统，一个数据架构。

Kafka外在表现很像消息系统，允许发布和订阅消息流，但是它和传统的消息系统有很大的差异，首先Kafka是个现代分布式系统，以集群的方式运行，可以自由伸缩。 其次Kafka可以按照要求存储数据，保存多久都可以，第三，流式处理将数据处理的层次提示到了新高度，消息系统只会传递数据，Kafka的流式处理能力可以让我们用很少的代码就能动态地处理派生流和数据集。所以Kafka不仅仅是个消息中间件，同时它是一个流平台，这个平台上可以发布和订阅数据流（Kafka的流有一个单独的包Stream的处理），并把他们保存起来进行处理，这个是Kafka作者的设计理念。

大数据领域，Kafka还可以看成实时版的Hadoop，但是还是有些区别，Hadoop可以存储和定期处理大量的数据文件，往往以TB计数，而Kafka可以存储和持续处理大型的数据流。Hadoop主要用在数据分析上，而Kafka因为低延迟，更适合于核心的业务应用上。所以国内的大公司一般会结合使用，比如京东在实时数据计算架构中就使用了到了Kafka，具体见《张开涛-海量数据下的应用系统架构实践》。常见的大数据处理框架：**storm****、****spark****、****Flink****、****(Blink** **阿里****)**

Kafka名字的由来：卡夫卡与法国作家马塞尔·普鲁斯特，爱尔兰作家詹姆斯·乔伊斯并称为西方现代主义文学的先驱和大师。《变形记》是卡夫卡的短篇代表作，是卡夫卡的艺术成就中的一座高峰，被认为是20世纪最伟大的小说作品之一（达到管理层的高度应该多看下人文相关的书籍，增长管理知识和人格魅力）。以kafka\_2.11-2.3.0版本为主，其余版本不予考虑，并且Kafka是scala语言写的，小众语言没有必要研究其源码，投入和产出比低，除非你的技术级别非常高或者需要去开发单独的消息中间件。

## **2、Kafka中的基本概念**  ##

### **2.1、消息和批次** ###

**消息—**Kafka里的数据单元，也就是我们一般消息中间件里的消息的概念（可以比作数据库中一条记录）。消息由字节数组组成，消息还可以包含键（可选元数据，也是字节数组），主要用于对消息选取分区。作为一个高效的消息系统，为了提高效率，消息可以被分批写入Kafka。

**批次—**就是一组消息，这些消息属于同一个主题和分区。如果只传递单个消息，会导致大量的网络开销，把消息分成批次传输可以减少这开销。但是，这个需要权衡（时间延迟和吞吐量之间），批次里包含的消息越多，单位时间内处理的消息就越多，单个消息的传输时间就越长（吞吐量高延时也高）。如果进行压缩，可以提升数据的传输和存储能力，但需要更多的计算处理。

对于Kafka来说，消息是晦涩难懂的字节数组，一般我们使用序列化和反序列化技术，格式常用的有JSON和XML，还有Avro（Hadoop开发的一款序列化框架），具体怎么使用依据自身的业务来定。

### **2.2、主题和分区**  ###

Kafka里的消息用主题进行分类（主题好比数据库中的表），主题下有可以被分为若干个分区（分表技术）。分区本质上是个提交日志文件，有新消息，这个消息就会以追加的方式写入分区（写文件的形式），然后用先入先出的顺序读取。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70][]

但是因为主题会有多个分区，所以在整个主题的范围内，是无法保证消息的顺序的，单个分区则可以保证。Kafka通过分区来实现数据冗余和伸缩性，因为分区可以分布在不同的服务器上，那就是说一个主题可以跨越多个服务器（这是Kafka高性能的一个原因，多台服务器的磁盘读写性能比单台更高）。

前面我们说Kafka可以看成一个流平台，很多时候，我们会把一个主题的数据看成一个流，不管有多少个分区。

### **2.3、生产者和消费者、偏移量、消费者群组**  ###

就是一般消息中间件里生产者和消费者的概念。一些其他的高级客户端API，像数据管道API和流式处理的Kafka Stream，都是使用了最基本的生产者和消费者作为内部组件，然后提供了高级功能。

生产者默认情况下把消息均衡分布到主题的所有分区上，如果需要指定分区，则需要使用消息里的消息键和分区器。消费者订阅主题，一个或者多个，并且按照消息的生成顺序读取。消费者通过检查所谓的偏移量来区分消息是否读取过。偏移量是一种元数据，一个不断递增的整数值，创建消息的时候，Kafka会把他加入消息。在一个主题中一个分区里，每个消息的偏移量是唯一的。每个分区最后读取的消息偏移量会保存到Zookeeper或者Kafka上，这样分区的消费者关闭或者重启，读取状态都不会丢失。

多个消费者可以构成一个消费者群组，怎么构成？共同读取一个主题的消费者们，就形成了一个群组，群组可以保证每个分区只被一个消费者使用。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1][]

消费者和分区之间的这种映射关系叫做消费者对分区的所有权关系，很明显一个分区只有一个消费者，而一个消费者可以有多个分区。一个分区只能被消费者群组中的一个消费者消费（不能重复消费），如果有一个消费者挂掉了，另外的消费者接上。

### **2.4、Broker和集群**  ###

一个独立的Kafka服务器叫Broker，broker的主要工作是，接收生产者的消息，设置偏移量，提交消息到磁盘保存；为消费者提供服务，响应请求，返回消息。在合适的硬件上，单个broker可以处理上千个分区和每秒百万级的消息量（要达到这个目的需要做操作系统调优和JVM调优）。

多个broker可以组成一个集群，每个集群中broker会选举出一个集群控制器。控制器会进行管理，包括将分区分配给broker和监控broker。集群里，一个分区从属于一个broker，这个broker被称为首领；但是分区可以被分配给多个broker，这个时候会发生分区复制，集群中Kafka内部一般使用管道技术进行高效的复制。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2][]

分区复制带来的好处是，提供了消息冗余，一旦首领broker失效，其他broker可以接管领导权；当然相关的消费者和生产者都要重新连接到新的首领上。

### **2.5、保留消息**  ###

在一定期限内保留消息是Kafka的一个重要特性，Kafka broker默认的保留策略是：要么保留一段时间（7天），要么保留一定大小（比如1个G）。到了限制，旧消息过期并删除，但是每个主题可以根据业务需求配置自己的保留策略（开发时要注意，Kafka不像Mysql之类的永久存储）。

## **3、为什么选择Kafka**  ##

### **3.1、优点** ###

*  多生产者和多消费者；
 *  基于磁盘的数据存储，换句话说，Kafka的数据天生就是持久化的。
 *  高伸缩性，Kafka一开始就被设计成一个具有灵活伸缩性的系统，对在线集群的伸缩丝毫不影响整体系统的可用性。
 *  高性能，结合横向扩展生产者、消费者和broker，Kafka可以轻松处理巨大的信息流（LinkedIn公司每天处理万亿级数据），同时保证亚秒级的消息延迟。

### **3.2、常见场景**  ###

**1）****活动跟踪**

跟踪网站用户和前端应用发生的交互，比如页面访问次数和点击，将这些信息作为消息发布到一个或者多个主题上，这样就可以根据这些数据为机器学习提供数据，更新搜素结果等等（头条、淘宝等总会推送你感兴趣的内容，其实在数据分析之前就已经做了活动跟踪）。

**2）****传递消息**

标准消息中间件的功能。

**3）****收集指标和日志**

收集应用程序和系统的度量监控指标，或者收集应用日志信息，通过Kafka路由到专门的日志搜索系统，比如ES（国内用得较多）。

**4）****提交日志**

收集其他系统的变动日志，比如数据库。可以把数据库的更新发布到Kafka上，应用通过监控事件流来接收数据库的实时更新，或者通过事件流将数据库的更新复制到远程系统。还可以当其他系统发生了崩溃，通过重放日志来恢复系统的状态（异地灾备）。

**5）****流处理**

操作实时数据流，进行统计、转换、复杂计算等等。随着大数据技术的不断发展和成熟，无论是传统企业还是互联网公司都已经不再满足于离线批处理，实时流处理的需求和重要性日益增长。近年来业界一直在探索实时流计算引擎和API，比如这几年火爆的Spark Streaming、Kafka Streaming、Beam和Flink，其中阿里双11会场展示的实时销售金额，就用的是流计算，是基于Flink，然后阿里在其上定制化的Blink。

# **二、Kafka的安装、管理和配置**  #

## **1、安装** ##

### **1.1、预备环境**  ###

Kafka是Java生态圈下的一员，用Scala编写，运行在Java虚拟机上，所以安装运行和普通的Java程序并没有什么区别。安装Kafka官方说法，Java环境推荐Java8，Kafka需要Zookeeper保存集群的元数据信息和消费者信息。Kafka一般会自带Zookeeper，但是从稳定性考虑，应该使用单独的Zookeeper，而且构建Zookeeper集群。

### **1.2、下载和安装Kafka**  ###

在http://kafka.apache.org/downloads上寻找合适的版本下载，我们这里选用的是kafka\_2.11-2.3.0，下载完成后解压到本地目录。

### **1.3、运行** ###

**1）启动****Zookeeper**，windows下kafka目录文件夹\\bin\\windows目录下，新建一个start\_zookeeper.bat运行文件，编辑文件内容输入：zookeeper-server-start.bat ../../config/zookeeper.properties

Cm命令框执行进入至kafka目录文件夹\\bin\\windows下，然后执行start\_zookeeper.bat，启动zookeeper成功后会弹出提示框，此框勿关闭。

**2）启动****Kafka****，**进入Kafka目录下的bin\\windows；新建一个start\_kafka.bat运行文件，编辑输入内容：kafka-server-start.bat ../../config/server.properties

Cmd命令框执行进入至kafka目录文件夹\\bin\\windows下，然后执行start\_kafka.bat，启动kafka成功后会弹出提示框，此框勿关闭。出现以下画面表示成功：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3][]

Linux下与此类似，进入bin后，执行对应的sh文件即可。

### **1.4、基本的操作和管理**  ###

\#\#列出所有主题

kafka-topics.bat --zookeeper localhost:2181 --list

\#\#列出所有主题的详细信息

kafka-topics.bat --zookeeper localhost:2181 --describe

\#\#创建主题 主题名 my-topic，1 副本，8 分区

kafka-topics.bat --zookeeper localhost:2181 --create --topic my-topic --replication-factor 1 --partitions 8

\#\#增加分区，注意：分区无法被删除

kafka-topics.bat --zookeeper localhost:2181 --alter --topic my-topic --partitions 16\#\#删除主题

kafka-topics.bat --zookeeper localhost:2181 --delete --topic my-topic

\#\#创建生产者（控制台）

kafka-console-producer.bat --broker-list localhost:9092 --topic my-topic

\#\#创建消费者（控制台）

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic my-topic --from-beginning

\#\#列出消费者群组（仅 Linux）

kafka-topics.sh --new-consumer --bootstrap-server localhost:9092 --list

\#\#列出消费者群组详细信息（仅 Linux）

kafka-topics.sh --new-consumer --bootstrap-server localhost:9092 --describe --group 群组名

### **1.5、Kafka的Linux下的运行** ###

步骤和window上是一样，先运行zookeeper再运行Kafka，运行目录在kafka目录文件夹\\bin目录下，使用sh脚本。

注意：Kafka运行占用的端口：Zookeeper默认端口2181，Kafka默认端口9092

## **2、Zookeeper数据查看工具ZooInspector** ##

Zookeeper作为常用的集群协调者组件被广泛应用，尤其是在大数据生态圈中；

Zookeeper集群存储各个节点信息，包括：Hadoop、Hbase、Storm、Kafka等等；

### **2.1、查询ZK数据的方式** ###

那如何查看Zookeeper中的数据呢，我们可以通过ZkCli.sh命令客户端查看，但是不太直观，因为ZK本身数据是以树型结构存储组织的，所以推荐一个实用的界面操作工具ZooInspector。

### **2.2、ZooInspector的使用** ###

1）下载：https://issues.apache.org/jira/secure/attachment/12436620/ZooInspector.zip

2）解压，进入目录ZooInspector\\build，运行：zookeeper-dev-ZooInspector.jar

运行结果：

java -jar zookeeper-dev-ZooInspector.jar  //执行成功后，会弹出java ui client

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4][]

## **3、Broker配置**  ##

配置文件放在Kafka目录下的config目录中，主要是server.properties文件

### **3.1、常规配置**  ###

**broker.id****：**在单机时无需修改，但在集群下部署时往往需要修改。它是个每一个broker在集群中的唯一表示，要求是正数。当该服务器的IP地址发生改变时，broker.id没有变化，则不会影响consumers的消息情况。

**Listeners****：**监听列表(以逗号分隔，不同的协议(如plaintext、trace、ssl、不同的IP和端口))hostname如果设置为0.0.0.0则绑定所有的网卡地址；如果hostname为空则绑定默认的网卡。如果没有配置则默认为 java.net.InetAddress.getCanonicalHostName()。如：PLAINTEXT://myhost:9092,TRACE://:9091或PLAINTEXT://

0.0.0.0:9092。

**zookeeper.connect****：**zookeeper集群的地址，可以是多个，多个之间用逗号分割（一组hostname:port/path列表，hostname是zk的机器名或IP，port是zk的端口，/path是可选zk的路径，如果不指定，默认使用根路径）。

**log.dirs****：**Kafka 把所有的消息都保存在磁盘上，存放这些数据的目录通过log.dirs指定。可以使用多路径，使用逗号分隔，如果是多路径，Kafka会根据“最少使用”原则，把同一个分区的日志片段保存到同一路径下，会往拥有最少数据分区的路径新增分区。

**num.recovery.threads.per.data.dir****：**每数据目录用于日志恢复启动和关闭时的线程数量。因为这些线程只是服务器启动（正常启动和崩溃后重启）和关闭时会用到。所以完全可以设置大量的线程来达到并行操作的目的。注意，这个参数指的是每个日志目录的线程数，比如本参数设置为8，而log.dirs设置为了三个路径，则总共会启动24个线程。

**auto.create.topics.enable****：**是否允许自动创建主题。如果设为true，那么produce生产者往主题写消息，consume消费者从主题读消息或者 fetch metadata（任意客户端向主题发送元数据请求时）一个不存在的主题时，就会自动创建，缺省为 true。

### **3.2、主题配置**  ###

新建主题的默认参数

**num.partitions****：**每个新建主题的分区个数(分区个数只能增加，不能减少)。这个参数一般要评估，比如，每秒钟要写入和读取1000M数据，如果现在每个消费者每秒钟可以处理50MB的数据，那么需要20 个分区，这样就可以让20个消费者同时读取这些分区，从而达到设计目标(一般经验，把分区大小限制在25G之内比较理想)。

**log.retention.hours****：**日志保存时间，默认为7天(168小时)，超过这个时间会清理数据。bytes和minutes无论哪个先达到都会触发。与此类似还有log.retention.minutes和log.retention.ms都设置的话，优先使用具有最小值的那个（提示：时间保留数据是通过检查磁盘上日志片段文件的最后修改时间来实现的，也就是最后修改时间是指日志片段的关闭时间，也就是文件里最后一个消息的时间戳）。

**log.retention.bytes****：**topic每个分区的最大文件大小，一个topic的大小限制=分区数\* log.retention.bytes，-1没有大小限制。log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除(注意如果是 log.retention.bytes先达到了，则是删除多出来的部分数据)，一般不推荐使用最大文件删除策略，而是推荐使用文件过期删除策略。

**log.segment.bytes****：**分区的日志存放在某个目录下诸多文件中，这些文件将分区的日志切分成一段一段的，我们称为日志片段。这个属性就是每个文件的最大尺寸；当尺寸达到这个数值时，就会关闭当前文件，并创建新文件。被关闭的文件就开始等待过期，默认为 1G。

如果一个主题每天只接受100MB的消息，那么根据默认设置，需要10天才能填满一个文件。而且因为日志片段在关闭之前，消息是不会过期的，所以如果log.retention.hours保持默认值的话，那么这个日志片段需要17天才过期。因为关闭日志片段需要10天，等待过期又需要7天。

**log.segment.ms****：**作用和log.segment.bytes类似，只不过判断依据是时间。同样的两个参数，以先到的为准，这个参数默认是不开启的。

**message.max.bytes****：**表示一个服务器能够接收处理的消息的最大字节数，注意这个值producer和consumer必须设置一致，且不要大于fetch.message.max.bytes属性的值(消费者能读取的最大消息，这个值应该大于或等于message.max.bytes)；该值默认是1000000字节，大概900KB~1MB。如果启动压缩，判断压缩后的值。这个值的大小对性能影响很大，值越大网络和IO的时间越长，还会增加磁盘写入的大小。Kafka设计的初衷是迅速处理短小的消息，一般10K大小的消息吞吐性能最好（LinkedIn的kafka性能测试）。

## **4、硬件配置对Kafka性能的影响** ##

为Kafka选择合适的硬件更像是一门艺术，就跟它的名字一样，我们分别从磁盘、内存、网络和CPU上来分析，确定了这些关注点，就可以在预算范围之内选择最优的硬件配置。

### **4.1、磁盘吞吐量/磁盘容量** ###

磁盘吞吐量（IOPS每秒的读写次数）会影响生产者的性能，因为生产者的消息必须被提交到服务器保存，大多数的客户端都会一直等待，直到至少有一个服务器确认消息已经成功提交为止。也就是说，磁盘写入速度越快，生成消息的延迟就越低（SSD固态贵，单个速度快，HDD机械偏移可以多买几个，设置多个目录加快速度，具体情况具体分析）磁盘容量的大小，则主要看需要保存的消息数量；如果每天收到1TB的数据，并保留7天，那么磁盘就需要7TB的数据。

### **4.2、内存** ###

Kafka本身并不需要太大内存，内存则主要是影响消费者性能。在大多数业务情况下，消费者消费的数据一般会从内存（页面缓存，从系统内存中分）中获取，这比在磁盘上读取肯定要快的多。一般来说运行Kafka的JVM不需要太多的内存，剩余的系统内存可以作为页面缓存，或者用来缓存正在使用的日志片段，所以我们一般Kafka不会同其他的重要应用系统部署在一台服务器上，因为他们需要共享页面缓存，这个会降低Kafka消费者的性能。

### **4.3、网络** ###

网络吞吐量决定了Kafka能够处理的最大数据流量，它和磁盘是制约Kafka拓展规模的主要因素。对于生产者、消费者写入数据和读取数据都要瓜分网络流量，同时做集群复制也非常消耗网络。

### **4.4、CPU** ###

Kafka对cpu的要求不高，主要是用在对消息解压和压缩上，所以cpu的性能不是在使用Kafka的首要考虑因素。

### **4.5、总结** ###

我们要为 Kafka 选择合适的硬件时，优先考虑存储，包括存储的大小，然后考虑生产者的性能（也就是磁盘的吞吐量），选好存储以后，再来选择CPU和内存就容易得多。网络的选择要根据业务上的情况来定，也是非常重要的一环。

# **三、Kafka的集群**  #

## **1、为何需要Kafka集群** ##

本地开发，一台Kafka足够使用。在实际生产中，集群可以跨服务器进行负载均衡，再则可以使用复制功能来避免单独故障造成的数据丢失，同时集群可以提供高可用性。

### **2、如何估算Kafka集群中Broker的数量** ###

要估量以下几个因素：

1）需要多少磁盘空间保留数据和每个broker上有多少空间可以用。比如，如果一个集群有10TB的数据需要保留，而每个broker可以存储2TB，那么至少需要5个broker；如果启用了数据复制，则还需要一倍的空间，那么这个集群需要10个 broker。

2）集群处理请求的能力。如果因为磁盘吞吐量和内存不足造成性能问题，可以通过扩展broker来解决。

### **3、Broker如何加入Kafka集群** ###

非常简单，只需要两个参数：

第一：配置zookeeper.connect

第二：为新增的broker设置一个集群内的唯一性id， Kafka中的集群是可以动态扩容的。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70]: /images/20230531/2092e262602841faaed58edd0108b7cb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 1]: /images/20230531/d993b6508ef641f892d3f6f51a5dbed5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 2]: /images/20230531/40047d07debb4f44ba90fd015110f06c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 3]: /images/20230531/625e98535c7a4c7d8f723d66ea0e9feb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3NjYxNDU4_size_16_color_FFFFFF_t_70 4]: /images/20230531/25229f67e5a44542bf7f0565d9e254ca.png