# Kafka架构与实战


<!--more-->


## 1. 概念和基本架构

### 1.1 介绍

Kafka是最初由Linkedin公司开发，是⼀个分布式、分区的、多副本的、多⽣产者、多订阅者，基于zookeeper协调的分布式⽇志系统（也可以当做MQ系统），常⻅可以⽤于web/nginx⽇志、访问⽇志，消息服务等等，Linkedin于2010年贡献给了Apache基⾦会并成为顶级开源项⽬。

**应用场景**: 日志收集系统, 消息系统

**主要设计目标如下:**

- 以时间复杂度为O(1)的⽅式提供消息持久化能⼒，即使对TB级以上数据也能保证常数时间的访问性能。
- ⾼吞吐率。即使在⾮常廉价的商⽤机器上也能做到单机⽀持每秒100K条消息的传输。
- ⽀持KafkaServer间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
- 同时⽀持离线数据处理和实时数据处理。
- ⽀持在线⽔平扩展

![img](https://img.zhaojq.top/20260727091958657.png "img")

> 1. Kafka在⼀个或多个可以跨越多个数据中⼼的服务器上作为集群运⾏。
> 2. Kafka集群中按照主题分类管理，⼀个主题可以有多个分区，⼀个分区可以有多个副本分区。
> 3. 每个记录由⼀个键，⼀个值和⼀个时间戳组成。

**有两种主要消息传递模式:**

- 点对点传递模式

- 发布-订阅模式

  > 大部分系统主要采用这种模式

**Kafka只有消息的拉取, 没有推送,可以通过轮询实现消息的推送**

**4个核心API**:

- Producer API: 允许应⽤程序将记录流发布到⼀个或多个Kafka主题
- Consumer API: 允许应⽤程序订阅⼀个或多个主题并处理为其⽣成的记录流
- Streams API：允许应⽤程序充当流处理器，使⽤⼀个或多个主题的输⼊流，并⽣成⼀个或多个输出主题的输出流，从⽽有效地将输⼊流转换为输出流。
- Connector API：允许构建和运⾏将Kafka主题连接到现有应⽤程序或数据系统的可重⽤⽣产者或使⽤者。例如，关系数据库的连接器可能会捕获对表的所有更改。

### 1.2 优势

- ⾼吞吐量：单机每秒处理⼏⼗上百万的消息量。即使存储了许多TB的消息，它也保持稳定的性能。
- ⾼性能：单节点⽀持上千个客户端，并保证零停机和零数据丢失。
- 持久化数据存储：将消息持久化到磁盘。通过将数据持久化到硬盘以及replication防⽌数据丢失。
  - 零拷贝
  - 顺序读, 顺序写
  - 利用linux的页缓存
- 分布式系统，易于向外扩展。所有的Producer、Broker和Consumer都会有多个，均为分布式的。⽆需停机即可扩展机器。多个Producer、Consumer可能是不同的应⽤。
- 可靠性-Kafka是分布式，分区，复制和容错的。
- 客户端状态维护：消息被处理的状态是在Consumer端维护，⽽不是由server端维护。当失败时能⾃动平衡
- ⽀持online和offline的场景。
- ⽀持多种客户端语⾔。Kafka⽀持Java、.NET、PHP、Python等多种语⾔。

### 1.3 应用场景

- 日志收集

- 消息系统

- 用户活动跟踪

  Kafka经常被⽤来记录Web⽤户或者App⽤户的各种活动，如浏览⽹⻚、搜索、点击等活动，这些活动信息被各个服务器发布到Kafka的Topic中，然后消费者通过订阅这些Topic来做实时的监控分析，亦可保存到数据库；

- 运营指标

  Kafka也经常⽤来记录运营监控数据。包括收集各种分布式应⽤的数据，⽣产各种操作的集中反馈，⽐如报警和报告；

- 流式处理

  ⽐如SparkStreaming和Storm。

### 1.4 基本架构

- 消息和批次

  Kafka的数据单元称为消息。可以把消息看成是数据库⾥的⼀个“数据⾏”或⼀条“记录”。消息由字节数组组成。

  消息有键，键也是⼀个字节数组。当消息以⼀种可控的⽅式写⼊不同的分区时，会⽤到键。

  为了提⾼效率，消息被分批写⼊Kafka。批次就是⼀组消息，这些消息属于同⼀个主题和分区。把消息分成批次可以减少⽹络开销。批次越⼤，单位时间内处理的消息就越多，单个消息的传输时间就越⻓。批次数据会被压缩，这样可以提升数据的传输和存储能⼒，但是需要更多的计算处理。

- 模式

  消息模式（schema）有许多可⽤的选项，以便于理解。如JSON和XML，但是它们缺乏强类型处理能⼒。Kafka的许多开发者喜欢使⽤**Apache Avro**。Avro提供了⼀种紧凑的序列化格式，**模式和消息体分开**。当模式发⽣变化时，不需要重新⽣成代码，它还⽀持强类型和模式进化，其版本既向前兼容，也向后兼容。

  数据格式的⼀致性对Kafka很重要，因为它消除了消息读写操作之间的耦合性

- 主题和分区

  Kafka的消息通过主题进⾏分类。主题可⽐是数据库的表或者⽂件系统⾥的⽂件夹。主题可以被分为若⼲分区，⼀个主题通过分区分布于Kafka集群中，提供了横向扩展的能⼒。![img_1](https://img.zhaojq.top/20260727091958838.png "img_1")

- 生产者和消费者

  ⽣产者创建消息。消费者消费消息。⼀个消息被发布到⼀个特定的主题上。⽣产者在默认情况下把消息均衡地分布到主题的所有分区上：

  - 直接指定消息的分区

  - 根据消息的key散列取模得出分区
  - 轮询指定分区。

  消费者通过偏移量来区分已经读过的消息，从⽽消费消息。消费者是消费组的⼀部分。消费组保证每个分区只能被⼀个消费者使⽤，避免重复消费。![img_2](https://img.zhaojq.top/20260727091959029.png "img_2")

- Broker和集群

  ⼀个独⽴的Kafka服务器称为broker。broker接收来⾃⽣产者的消息，为消息设置偏移量，并提交消息到磁盘保存。broker为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘上的消息。**单个broker**可以轻松处理**数千个分区**以及**每秒百万级**的消息量。![img_3](https://img.zhaojq.top/20260727091959199.png "img_3")

  每个集群都有⼀个broker是集群控制器（⾃动从集群的活跃成员中选举出来）。控制器负责管理⼯作：

  - 将分区分配给broker 
  - 监控broker

  集群中⼀个分区属于⼀个broker，该broker称为**分区⾸领**。⼀个分区可以分配给多个broker，此时会发⽣分区复制。分区的复制提供了消息冗余，⾼可⽤。副本分区不负责处理消息的读写。

### 1.5 核心概念

- Producer

  ⽣产者创建消息。该⻆⾊将消息发布到Kafka的topic中。broker接收到⽣产者发送的消息后，broker将该消息追加到当前⽤于追加数据的segment文件中

  **⼀般情况下，⼀个消息会被发布到⼀个特定的主题上。**

  - 默认情况下通过轮询把消息均衡地分布到主题的所有分区上。
  - 在某些情况下，⽣产者会把消息直接写到指定的分区。这通常是通过消息键和分区器来实现的，分区器为键⽣成⼀个散列值，并将其映射到指定的分区上。这样可以保证包含同⼀个键的消息会被写到同⼀个分区上。
  - ⽣产者也可以使⽤⾃定义的分区器，根据不同的业务规则将消息映射到分区

- Consumer

  消费者读取消息。

  - 消费者订阅⼀个或多个主题，并按照消息⽣成的顺序读取它们。
  - 消费者通过检查消息的偏移量来区分已经读取过的消息。偏移量是另⼀种元数据，它是⼀个**不断递增的整数值**，在创建消息时，Kafka会把它添加到消息⾥。在给定的分区⾥，每个消息的偏移量都是唯⼀的。消费者把每个分区最后读取的消息偏移量保存在Zookeeper或Kafka上，如果消费者关闭或重启，它的读取状态不会丢失。
  - 消费者是消费组的⼀部分。群组保证每个分区只能被⼀个消费者使⽤。
  - 如果⼀个消费者失效，消费组⾥的其他消费者可以接管失效消费者的⼯作，再平衡，分区重新分配。

- Broker

  ⼀个独⽴的Kafka服务器被称为broker。broker为消费者提供服务，对读取分区的请求作出响应，返回已经提交到磁盘上的消息。

  - 如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的⼀个partition。
  - 如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的⼀个partition，剩下的M个broker不存储该topic的partition数据。
  - 如果某topic有N个partition，集群中broker数⽬少于N个，那么⼀个broker存储该topic的⼀个或多个partition。在实际⽣产环境中，尽量避免这种情况的发⽣，这种情况容易导致Kafka集群数据不均衡。

  broker是集群的组成部分。每个集群都有⼀个broker同时充当了集群控制器的⻆⾊（⾃动从集群的活跃成员中选举出来）。

  控制器负责管理⼯作，包括将分区分配给broker和监控broker。

  在集群中，⼀个分区从属于⼀个broker，该broker被称为分区的⾸领。![img_4](https://img.zhaojq.top/20260727091959368.png "img_4")

- Topic

  每条发布到Kafka集群的消息都有⼀个类别，这个类别被称为Topic。物理上不同Topic的消息分开存储。主题就好⽐数据库的表，尤其是分库分表之后的逻辑表。

- Partition

  - 主题可以被分为若⼲个分区，⼀个分区就是⼀个提交⽇志。
  - 消息以追加的⽅式写⼊分区，然后以先⼊先出的顺序读取。
  - ⽆法在整个主题范围内保证消息的顺序，但可以保证消息在单个分区内的顺序。4.Kafka通过分区来实现数据冗余和伸缩性。
  - 在需要严格保证消息的消费顺序的场景下，需要将partition数⽬设为1。
  - ![img_5](https://img.zhaojq.top/20260727091959577.png "img_5")

- Replicas

  Kafka使⽤主题来组织数据，每个主题被分为若⼲个分区，每个分区有多个副本。那些副本被保存在broker上，每个broker可以保存成百上千个属于不同主题和分区的副本。

  - 首领副本

    每个分区都有⼀个⾸领副本。为了保证⼀致性，所有⽣产者请求和消费者请求都会经过这个副本

  - 跟随者副本

    ⾸领以外的副本都是跟随者副本。跟随者副本不处理来⾃客户端的请求，它们唯⼀的任务就是从⾸领那⾥复制消息，保持与⾸领⼀致的状态。如果⾸领发⽣崩溃，其中的⼀个跟随者会被提升为新⾸领。

- Offset

  - ⽣产者Offset 

    消息写⼊的时候，每⼀个分区都有⼀个offset，这个offset就是⽣产者的offset，同时也是这个分区的最新最⼤的offset。有些时候没有指定某⼀个分区的offset，这个⼯作kafka帮我们完成。![img_6](https://img.zhaojq.top/20260727091959763.png "img_6")

  - 消费者Offset

    ![img_7](https://img.zhaojq.top/20260727091959922.png "img_7")

    这是某⼀个分区的offset情况，⽣产者写⼊的offset是最新最⼤的值是12，⽽当ConsumerA进⾏消费时，从0开始消费，⼀直消费到了9，消费者的offset就记录在9，ConsumerB就纪录在了11。等下⼀次他们再来消费时，他们可以选择接着上⼀次的位置消费，当然也可以选择从头消费，或者跳到最近的记录并从“现在”开始消费。

- 副本

  Kafka通过副本保证⾼可⽤。副本分为**⾸领副本(Leader)**和**跟随者副本(Follower)**。

  跟随者副本包括**同步副本**和**不同步副本**，在发⽣⾸领副本切换的时候，只有同步副本可以切换为⾸领副本。

  - AR

    分区中的所有副本统称为AR(Assigned Repllicas) AR=ISR+OSR 

  - ISR

    所有与leader副本保持⼀定程度同步的副本（包括Leader）组成ISR（In-SyncReplicas），ISR集合是AR集合中的⼀个⼦集。消息会先发送到leader副本，然后follower副本才能从leader副本中拉取消息进⾏同步，同步期间内follower副本相对于leader副本⽽⾔会有⼀定程度的滞后。前⾯所说的“⼀定程度”是指可以忍受的滞后范围，这个范围可以通过参数进⾏配置。

  - OSR

    与leader副本同步滞后过多的副本（不包括leader）副本，组成OSR(Out-SyncRelipcas)。在正常情况下，所有的follower副本都应该与leader副本保持⼀定程度的同步，即AR=ISR,OSR集合为空。

  - HW

    HW是High Watermak的缩写，俗称⾼⽔位，它表示了⼀个特定消息的偏移量（offset），消费之只能拉取到这个offset之前的消息。

  - LEO

    LEO是LogEndOffset的缩写，它表示了当前⽇志⽂件中下⼀条待写⼊消息的offset。

## 2. 安装与配置

### 2.1 java环境安装

安装jdk8即可, 不再做演示了, 记得配置环境变量

```properties
export JAVA_HOME=/usr/local/jdk1.8.0_231
export JRE_HOME=/usr/local/jdk1.8.0_231/jre
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib:$CLASSPATH
export PATH=$JAVA_HOME/bin:$PATH
```

### 2.2 zookeeper的安装

- 下载稳定版zookeeper

  ```http
   wget https://dlcdn.apache.org/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz
  ```
  
- 解压并进入conf目录,配置以下两个参数
  
  记得创建对应的目录
  
  ```properties
  dataDir=/tmp/zookeeper
  dataLogDir=/tmp/zookeeper/logs
  ```
  
- 环境变量配置

  ```sh
  export ZOOKEEPER_PREFIX=/home/apache-zookeeper-3.7.1-bin #根据自己的安装目录
  export PATH=$PATH:$ZOOKEEPER_PREFIX/bin
  ```

- 生效配置

  ```sh
  source /etc/profile
  ```

- 验证

  ```sh
  zkServer.sh status
  ```

### 2.3 kafka的安装与配置

- 下载

  ```http
  wget https://archive.apache.org/dist/kafka/1.0.0/kafka_2.12-1.0.0.tgz
  ```

- 解压并配置环境变量

  ```properties
  tar -zxf kafka_2.12-1.0.0.tgz
  vim /etc/profile 
  
  export KAFKA_HOME=/home/kafka_2.12-1.0.0 #解压目录
  export PATH=$PATH:$KAFKA_HOME/bin
  
  #使生效
  source /etc/profile
  ```

- 配置config目录下的server.properties

  ```properties
  #Kafka连接Zookeeper的地址，此处使⽤本地启动的Zookeeper实例，连接地址是localhost:2181，
  #后⾯的myKafka 是Kafka在Zookeeper中的根节点路径
  zookeeper.connect=localhost:2181/myKafka
  
  #配置kafka连接
  #内网
  listeners=PLAINTEXT://192.168.0.237:9092 
  #外网,此处配置外网ip才能外网连接
  advertised.listeners=PLAINTEXT://114.115.171.220:9092
  ```

- 启动zookeeper并检查状态

  ```sh
  zkServer.sh start
  zkServer.sh status
  ```

  ![img_8](https://img.zhaojq.top/20260727092000105.png "img_8")

- 启动kafka

  ```sh
  # 启动命令  配置文件路径
  kafka-server-start.sh kafka_2.12-1.0.0/config/server.properties
  # 后台启动
  kafka-server-start.sh -daemon kafka_2.12-1.0.0/config/server.properties
  # 出现[KafkaServer id=0] started (kafka.server.KafkaServer)启动成功
  ```

- 查看zookeeper的节点

  ![img_9](https://img.zhaojq.top/20260727092000272.png "img_9")

### 2.4 生产与消费

- kafka-topics.sh用于管理主题

  ```sh
  # 列出现有的主题
  kafka-topics.sh --list --zookeeper localhost:2181/myKafka
  # 创建主题，该主题包含⼀个分区，该分区为Leader分区，它没有Follower分区副本
  kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_1 --partitions 1 --replication-factor 1
  # 查看指定主题的详细信息
  kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_1
  # 删除指定主题
  kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_1
  ```

- kafka-console-producer.sh⽤于⽣产消息

  ```sh
  # 开启⽣产者
  kafka-console-producer.sh --topic topic_1 --broker-list localhost:9092
  ```

- kafka-console-consumer.sh⽤于消费消息

  ```sh
  # 开启消费者
  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1
  
  # 开启消费者⽅式⼆，从头消费，不按照偏移量消费
  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1 --from-beginning
  ```

## 3. kafka开发实战

### 3.1 消息的发送与接收

![img_10](https://img.zhaojq.top/20260727092000466.png "img_10")

生产者主要的对象有:

- KafkaProducer

  发送消息的类

- ProducerRecord

  封装kafka的消息

KafkaProducer创建需要指定的参数有:

| 参数              | 说明                                                         |
| ----------------- | ------------------------------------------------------------ |
| bootstrap.servers | 配置⽣产者如何与broker建⽴连接。该参数设置的是初始化参数。如果⽣产者需要连接的是Kafka集群，则这⾥配置集群中⼏个broker的地址，⽽不是全部，当⽣产者连接上此处指定的broker之后，在通过该连接发现集群中的其他节点。 |
| key.serializer    | 要发送信息的key数据的序列化类。设置的时候可以写类名，也可以使⽤该类的Class对象。 |
| value.serializer  | 要发送消息的alue数据的序列化类。设置的时候可以写类名，也可以使⽤该类的Class对象。 |
| acks              | 默认值：all。<br/>acks=0：<br/>⽣产者不等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。<br/>该情形不能保证broker是否真的收到了消息，retries配置也不会⽣效。发送的消息的返回的消息偏移量永远是-1。<br/>acks=1 <br/>表示消息只需要写到主分区即可，然后就响应客户端，⽽不等待副本分区的确认。<br/>在该情形下，如果主分区收到消息确认之后就宕机了，⽽副本分区还没来得及同步该消息，则该消息丢失。<br/>acks=all <br/>⾸领分区会等待所有的ISR副本分区确认记录。<br/>该处理保证了只要有⼀个ISR副本分区存活，消息就不会丢失。这是Kafka最强的可靠性保证，等效于acks=-1 |
| retries           | retries重试次数<br/>当消息发送出现错误的时候，系统会重发消息。<br/>跟客户端收到错误时重发⼀样。<br/>如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送成功了 |

其他参数可以从 org.apache.kafka.clients.producer.ProducerConfig 中找到。我们后⾯的内容会介绍到。

**示例代码: kafka-demo#kafka-client-demo#demo01**(https://gitee.com/ixinglan/kafka-demo.git)

### 3.2 Springboot Kafka

**示例代码: kafka-demo#springboot-kafka-demo**(https://gitee.com/ixinglan/kafka-demo.git)

## 4. 服务端参数配置

server.properties⽂件中的配置

### 4.1 zookeeper.connect

该参数⽤于配置Kafka要连接的Zookeeper/集群的地址, 例:

```properties
zookeeper.connect=node2:2181,node3:2181,node4:2181/myKafka
```

### 4.2 listeners

⽤于指定当前Broker向外发布服务的地址和端⼝

 advertised.listeners 配合，⽤于做内外⽹隔离。

**内外网隔离配置:**

- **listener.security.protocol.map**

  监听器名称和安全协议的映射配置, ⽐如，可以将内外⽹隔离，即使它们都使⽤SSL

  `listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL`

  每个监听器的名称只能在map中出现⼀次。

- **listeners**

  ⽤于配置broker监听的URI以及监听器名称列表，使⽤逗号隔开多个URI及监听器名称

  如果监听器名称代表的不是安全协议，必须配置listener.security.protocol.map

  每个监听器必须使⽤不同的⽹络端⼝。

- **advertised.listeners**

  需要将该地址发布到zookeeper供客户端使⽤，如果客户端使⽤的地址与listeners配置不同

  可以在zookeeper的` get /myKafka/brokers/ids/<broker.id>` 中找到

  如果不设置此条⽬，就使⽤listeners的配置。跟listeners不同，该条⽬不能使⽤0.0.0.0⽹络端⼝。

  advertised.listeners的地址必须是listeners中配置的或配置的⼀部分。

- **inter.broker.listener.name**

  ⽤于配置broker之间通信使⽤的监听器名称，该名称必须在advertised.listeners列表中

  如: `inter.broker.listener.name=EXTERNAL`

配置示例

```properties
#即名称与协议的映射关系
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
#下面INTERNAL相当于给内部管理用,EXTERNAL会通过advertised.listeners配置暴露给客户端用
listeners=INTERNAL://内网ip:9092,EXTERNAL://外网ip:9093
#暴露给zk给客户端使用
advertised.listeners=EXTERNAL://外网ip:9093
#在advertised.listeners中取
inter.broker.listener.name=EXTERNAL
```

### 4.3 broker.id

该属性⽤于唯⼀标记⼀个Kafka的Broker，它的值是⼀个任意integer值。

当Kafka以分布式集群运⾏的时候，尤为重要。

最好该值跟该Broker所在的物理主机有关的，如主机名为 host1.aaa.com ，则 broker.id=1 ，如果主机名为 192.168.100.101 ，则 broker.id=101 等等。

### 4.4 log.dir

通过该属性的值，指定Kafka在磁盘上保存消息的⽇志⽚段的⽬录, 它是⼀组⽤逗号分隔的本地⽂件系统路径。

如果指定了多个路径，那么broker 会根据“最少使⽤”原则，把同⼀个分区的⽇志⽚段保存到同⼀个路径下。

broker 会往拥有最少数⽬分区的路径新增分区，⽽不是往拥有最⼩磁盘空间的路径新增分区