一文详细解析Kakfa：生产者

- 日理万妓 2024-03-26 21:14 80阅读 0赞

## 前言 ##

生产者负责向Kakfa发送消息。扮演的是一个消息投递的角色。为了保证消息能够顺利安全地发送到Kafka broker里面，Kafka对生产者精心设计许多机制。

本篇介绍Kafka中生产者的一些基础配置和一些生产者的机制。

看完这篇文章你将会收获：

![format_png][]

## 生产者 ##

### 消息发送的流程 ###

![format_png 1][]

> 消息的构成：

*  **Topic**：主题
 *  **Partition**：分区
 *  **key**：如果key有值，分区器会根据它进行分区
 *  **valuea**：消息内容

`ProducerRecord`对象包含**目标主题（Topic）** 和要发送的内容。生产者需要把**键和值对象序列化成字节数组**。

然后数据会序**列化器**解析，得到解析后的数据，再扔给分区器。

如果在ProducerRecord对象里面指明了分区，那么分区器就不多做处理。如果没有指明，那就靠分区器**根据对象的键**来选择一个分区。

选择好分区之后，生产者就知道往哪个主题和分区发送这条记录。这条记录会被**追加**到一个记录批次里面，**一个批次里面的所有消息会被发送到相同的主题和分区上**。这部分工作是有**一个独立的线程**负责把这些记录批次发送到相应的broker上。

服务器收到这些消息时，就会返回一个响应。

如果消息写入Kafka成功，就返回一个`RecordMetaData`**元数据对象**，它包含了**主题**和**分区信息**，以及**记录在分区的偏移量**。

如果消息写入Kafka失败，就会返回一个错误，**生产者收到错误之后会尝试重新发送消息**，几次之后如果还是实不行，就返回错误信息。 **（最大努力交付）**

### 生产者配置 ###

> 要想往Kafka写入消息，第一步就是要创建一个生产者对象，并且设置一些配置属性

#### 必选配置 ####

下面列举几个Kafka生产者有`必选属性`：

bootstrap.servers

指定broker的地址清单，地址的格式为`host:sport`。清单里面不需要包含所有broker地址，**生产者会从给定的broker里面找到其他broker的信息**。（建议是提供两个broker，一旦其中一个宕机，另外一个顶上去连接到集群）

key.serializer

broker希望接收到的键值都是**字节数组**。生产者接口允许使用**参数化类型（Java里面叫泛型）** ，因此可以把java对象作为键或值发送给broker。默认提供了`ByteArraySerializer`、`StringSerializer`和`IntegerSerializer`。

value.serializer

跟前者一样，如果key、value都是同一个类型，那么就跟key.serializer使用同一个序列化器即可。

#### 可选配置 ####

acks

> 满足acks的设置，才代表生产者成功写入消息

![format_png 2][]

*  `acks=0`：代表生产者不需要任何服务器的响应。这种情况下，消息丢了就丢了，生产者自己也不知道。适合高吞吐量，但是消息并不重要的场景。
 *  `acks=1`：只要**首领副本**收到消息，生产者就会收到来自服务器的**成功响应**。如果消息无法到达首领节点（比如首领节点崩溃，新的首领还没有被选举出来等），生产者就会收到一个**错误响应**。生产者会尝试重发消息。**此时的吞吐量取决于是同步发送模式，还是异步发送模式**。
 *  `acks=all`：只有当**所有副本**都收到消息，生产者才会收到来自服务器的一个成功响应。

buffer.memory

> 生产者内存缓冲池大小

用来设置**生产者内存缓冲区的大小**，生产者用它缓冲要发送到服务器的消息，

compression.type

> 压缩算法，指的是**消息被发送到broker之前使用哪一种压缩算法进行压缩**。

消息在发送到broker之前，要走网络，那么就意味着需要消耗带宽。如果我们这个消息大小太大，性能上就很不好。所以我们一般会采取某种压缩算法来压缩消息。

*  `snappy`：**snappy**是谷歌研发的，它占用较少的CPU，却可以得到较好的性能和压缩比，**如果是比较关注性能和互联网带宽（比如直播系统）** 可以使用这种算法。
 *  `gzip`：**gzip**一般会占用较多的CPU，但是会**提高更高的压缩比**，如果带宽有限，可以采用这种算法。特别的gzip对文本类型压缩有特别好的效果。
 *  `lz4`：**lz4**则是**追求压缩解压速度**，他的压缩比并不是很好。如果网络带宽条件比较好，可以采用这种压缩算法。

retries

> 生产者重发消息的次数

字面意思，生产者重发消息的次数，超过了之后就会抛出一个重试异常。

batch.size

> 批次的大小

![format_png 3][]

**按照字节数计算，不是按照消息个数来计算**。当批次被填满的时候，批次里面的所有消息被发送出去。有时候也不需要等到被填满才发送，半满，甚至只包含1个消息的批次也有可能被发送。

*  批次太大：占内存。

*  批次太小：频繁发送消息，带来额外的开销。

linger.ms

> 发送批次前等待下一个消息加入的时间，也就是**批次的停车等待时间**

KafkaProducer会在**批次被填满**或者**linger.ms达到上限**时把批次发送出去。

默认情况下，只要有可用的线程，生产者就会把消息发送出去。

max.in.flight.requests.per.connection

> 服务器响应之前能接受的消息数

生产者在收到服务器响应之前，可以发送多少个消息，它的值越高，约占内存，不过吞吐量也越高。

**设置为1，则可以保证消息是按照发送顺序写入服务器的，即便是发送了重试**。

但是其实没必要阿，**Kafka是可以保证同一个分区**里的消息是有序得。只要生产者按照一定顺序发送消息，broker就会按照这个顺序将他们写入同一个分区。消费者消费的时候也是按顺序消费的。

### 消息发送方式 ###

实例化生产者对象之后，就可以向Kafka发送消息了，发送消息有3种方式。消息先是被放进**缓冲区**，然后使用单独的线程发送到服务端。

#### 发送并忘记（fire-and-forget） ####

> 发送给服务器之后，不关心他是否成功到达，大多情况下，消息会正常到达，因为Kafka是高可用，并且关键是生产者会**自动尝试重发**，有一定几率会丢失消息。

ProducerRecord<String, String> record =
         new ProducerRecord<>("CustomerCountry", "Precision Products",
                              "France"); 
     try {
         producer.send(record); 
     } catch (Exception e) {
         e.printStackTrace();
     }
    复制代码

#### 同步发送 ####

> 使用`send()`方法发送消息，他会返回一个**Future对象**，通过调用get()方法进行等待，就可以知道消息是否发送成功。

同步，指的是发送给Kafka之后，我这边还需要有一个**发送服务端等待服务请响应过程**，通过调用那个Future对象的get方法来进行处理后续逻辑。

ProducerRecord<String, String> record =
         new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
     try {
         producer.send(record).get(); 
     } catch (Exception e) {
         e.printStackTrace(); 
     }
    复制代码

❌KafKaProducer一般会发生两种类型`错误`：

*  一类是**可重试错误**，这类错误可以通过重发消息来解决。比如对于**连接错误**，可以通过再次建立连接来解决。“`无主`”错误（**no-leader**）则是可以通过重新为分区重新选举首领来解决。如果多次都无法解决问题，则会抛出一个**重试异常**。
 *  另外一类为**无法通过重试解决**，比如消息太大的异常，这类异常KafkaProducer不会进行任何重试，直接抛出异常。

#### 异步发送 ####

> 调用`send()`方法，指定一个**回调函数**，服务器在返回响应时调用该函数。

异步，指的是我不需要去处理消息等待过程，我们通过指定一个回调函数，让Kafka那边收到消息之后调用这个回调函数即可。

大多数情况，我们不需要等待响应，虽然说，Kafka会把目标主题、分区信息和消息的偏移量发送过来，但是对于发送端的应用程序来说，并不是必须的。**不过我们遇到消息发送失败的时候，需要抛出异常，记录错误日志，或者把消息写入“错误消息”文件以便日后分析**。

private class DemoProducerCallback implements Callback {
         @Override
         public void onCompletion(RecordMetadata recordMetadata, Exception e) {
             if (e != null) {
                 e.printStackTrace(); 
             }
         }
     }
     
     ProducerRecord<String, String> record =
         new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA"); 
     producer.send(record, new DemoProducerCallback());
    复制代码

**优点就是，我们可以对异常情况进行处理。**

### 序列化器 ###

> 之所以为什么需要使用序列化器，是因为我们有许多客户端，**每个客户端传给Kafka的键值类型都是可以不同的**，所以需要客户端指明使用的序列化器，**让Kafka知道怎么去解析数据**。

推荐的使用方案是：`Avro`。（Hadoop也是使用这个）

Kafka有自己提供的**默认序列化器（byteArray、String、Integer等）** ，也可以我们用户自己去**自定义序列化器**。

#### 自定义序列化器 ####

我们可以用业内常见的序列化框架，如`Avro`、`Thrift`或是`Protobuf`。**不建议使用我们自定义序列化器**。

因为当我们需要更新我们这个对象，新增或删除某一个字段，那么这个自定义序列化器就要发生改变。可能不止一个客户端用这个对象，那么所有的客户端序列化器都要跟着改。**非常不利于扩展**。

> 如何写一个自定义序列化器？

首先我们要知道Kafka接受的是**字节数组**，所以我们只需要**把对象序列为字节数组返回出去即可**。

#### 使用Avro序列化 ####

> **Apache Avro**是一种与编程语言无关的序列化格式。

Avro数据通过与语言无关的**schema**来定义。

**schema**通过**JSON**来描述，数据被序列化成**二进制文件**或者**JSON文件**。（一般都是用二进制文件）

Avro在读、写数据都需要用到**schema**。**schema存在于Avro的数据文件中**。

Avro的特点在于，负责写消息的应用程序如果使用了新的**schema**（比如新增或删除某一个字段），**负责读消息的应用程序可以继续处理消息而无需做任何改动**。所以它非常适合Kafka。

**注意**：虽然负责读消息的应用程序不需要改schema，**但是它还是会读不到最新的字段，只不过他不会返回错误或者异常**，读取到的是null。

#### 在Kafka中使用Avro ####

> Avro的数据文件包含了整个schema，虽然说这种开销对是可接受的，但是Kafka有许多消息，**那么每个消息所带来的负担是不可忽视的**。如何解决呢？Kafka有他自己的一套东西。

Kafka采取的是使用**schema注册表**来达到目标，这个不是Kafka自己实现的，需要借助外部来实现。我们用的是**Confluent Schema Registry**。

原理就是非常简单，原先Avro数据里面放schema，现在不放了，**改为将所有的schema放在同一个地方存着（这个地方就是schema注册表）** ，然后解析的时候根据Avro数据里的schema标识符去自己拉取schema下来。因此我们在发送消息的时候需要注册schema到schema注册表里面，然后塞入一个shcema标识符即可。

本质上：就是一个**拉模式（poll）** ，或者说是**读扩散**。

### 分区 ###

> 一个主题下，有一个或多个分区，**在同一个分区内，消息是具有顺序性的**。

ProducerRecord对象包含了目标主题、键、值。

**键**：可以设置为默认的**null**，不过大多数应用程序会用到null。键主要有2个用途：

*  可以作为消息的附加消息。
 *  也可以**决定消息该写到主题的哪个分区**。具有**相同键**的消息可以被写到**同一个分区**。

使用null作为键值，并且使用了默认的分区器，那么记录就会被**随机地发送到主题内各个可用的分区上**。

如果键不为空，并且使用了默认的分区器。Kafka会**对键进行散列**，然后根据散列后的结果，把消息映射到特定的分区上。所以同一个键总是会被映射到同一个分区上。

**关键一点是**：这边散列的分区是所有的分区，并不是可用的分区。所以有可能映射到不可用的分区。但是这种情况很少发生，并且Kafka具有**复制功能**和**可用性**。

一般我们是不会轻易改变主题分区数量。因为一旦改变了，那么**所有的映射关系都会发生变化**。有可能同一个键的数据会被映射到不同的分区。所以如果想要通过键来映射分区，那么最好在创建主题的时候就把分区规划好。

> 原则上是：永远不要新增分区

#### 分区器 ####

默认分区器就是上面所讨论的情况，他是使用次数最频繁、最常用的分区器。**它采用的是散列分区这么一个策略**。

有些时候，我们的业务需求，需要我们对数据指定分区，并且支持**单独性的分区**，比如某个大客户的账号记录我们想要单独分配到某个分区。

#### 黏性分区：StickyPartitioning Strategy ####

> 0.10版本之后的kafka实现了**黏性分区策略**，实现生产者发送数据**分块**优化。

我们知道，往Kafka发送消息，broker并不会立刻接收到消息。Kafka有**按量**和**按时**进行一个**Batch批次**的消息发送。

从这个设计上来说，我们当然希望是**消息尽可能填满一个批次**，这样是最赚的。

实际上，决定**Batch**如何形成的一个因素是分区策略（**Partition Strategy**）。

在`Kafka2.4`版本之前，采用的默认分区策略是**轮询**（Round-Robin），（既没有指定**partition**，又没有指定**key**的情况下，**如果多条消息不是被发送到相同的分区**，那么他们就不能被放到一个batch里）

所以这样就会造成一个大的Batch被拆分成多个小Batch。因此社区推出了一种新的分区策略**黏性分区**。

黏性分区：**会随机选择一个分区并尽可能地坚持使用该分区**，代表黏住这个分区。

`好处`：**显著地降低给消息指定分区过程中地延时**，有助于改进消息批处理，减少延迟，并减少broker的负载。

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/071e7c9ac5394c40a494178adc4ba931.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/998ec75dfdbc4720a0d0ba25919840d5.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/d32f503eeb164b628adaf2ee04f3a971.png
[format_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/bd2b8b0a424547818fdb240ed10161fb.png