kafka生产者你不得不知的那些事儿

前言

生产者流程

kafka生产者最重要的就是消息发送的整个流程，我们来看下究竟是怎么一回事把。

main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

kafkaProducer

RecordAccumulator

拦截器

序列化器： 用于在网络传输中将数据序列化为字节流进行传输，保证数据不会丢失。
分区器： 用于按照一定的规则将数据分发到不同的kafka broker节点中

Sender 线程负责从 RecordAccumulator 获取消息并将其发送到 Kafka 中。

RecordAccumulator

Sender

RecordAccumulator 缓存的大小可以通过生产者客户端参数 buffer.memory 配置，默认值为 33554432B，即 32M。
主线程中发送过来的消息都会被迫加到 RecordAccumulator 的某个双端队列（ Deque ）中，RecordAccumulator 内部为每个分区都维护了一个双端队列，即 Deque<ProducerBatch>, 消息写入缓存时，追加到双端队列的尾部。
Sender 读取消息时，从双端队列的头部读取。ProducerBatch 是指一个消息批次；与此同时，会将较小的 ProducerBatch 凑成一个较大 ProducerBatch，也可以减少网络请求的次数以提升整体的吞吐量。ProducerBatch 大小可以通过batch.size 控制，默认16kb。
Sender 线程会在有数据积累到batch.size,默认16kb，或者如果数据迟迟未达到batch.size，Sender线程等待linger.ms设置的时间到了之后就会获取数据。linger.ms单位ms，默认值是0ms，表示没有延迟。

Sender 从 RecordAccumulator 获取缓存的消息之后，会将数据封装成网络请求<Node,Request> 的形式，这样就可以将 Request 请求发往各个 Node 了。
请求在从 sender 线程发往 Kafka 之前还会保存到 InFlightRequests 中，它的主要作用是缓存了已经发出去但还没有收到服务端响应的请求。InFlightRequests默认每个分区下最多缓存5个请求，可以通过配置参数为max.in.flight.request.per. connection修改。
请求Request通过通道Selector发送到kafka节点。
发送后，需要等待kafka的应答机制，取决于配置项acks.

1：生产者发送过来的数据，Leader 收到数据后应答。
-1（all）：生产者发送过来的数据，Leader和副本节点收齐数据后应答。默认值是-1，-1 和all 是等价的。

Request请求接受到kafka的响应结果，如果成功的话，从InFlightRequests 清除请求，否则的话需要进行重发操作，可以通过配置项retries决定，当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。
清理消息累加器RecordAccumulator 中的数据。

生产者重要参数

bootstrap.servers

生产者连接集群所需的 broker 地址清单。例如hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需要所有的 broker 地址，因为生产者从给定的 broker里查找到其他 broker 信息。

key.serializer 和 value.serializer

buffer.memory

RecordAccumulator 缓冲区总大小，默认 32m。

batch.size

linger.ms

如果数据迟迟未达到 batch.size，kafka等待这个时间之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。

max.request.size

compression.type

这个参数用来指定消息的压缩方式，默认值为“none "，即默认情况下，消息不会被压缩。该参数还可以配置为 "gzip"，"snappy" 和 "lz4"。对消息进行压缩可以极大地减少网络传输、降低网络 I/O，从而提高整体的性能。消息压缩是一种以时间换空间的优化方式，如果对时延有一定的要求，则不推荐对消息进行压缩；

acks

acks的值为0，1和-1或者all。

Producer

1表示Producer 往集群发送数据只要 Leader 成功写入消息就可以发送下一条，只确保 Leader 接收成功。
-1 或 all表示Producer 往集群发送数据需要所有的 ISR Follower 都完成从 Leader 的同步才会发送下一条，确保Leader 发送成功和所有的副本都成功接收。安全性最高，但是效率最低。

max.in.flight.requests.per.connection

retries和retry.backoff.ms

当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。在kafka3.4.0默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置max.in.flight.requests.per.connection=1否则在重试此失败消息的时候，其他的消息可能发送成功了。另外retry.backoff.ms控制两次重试之间的时间间隔，默认是 100ms。

https://kafka.apache.org/documentation/#producerconfigs。

生产者发送消息API

生产者发送demo

（1）配置生产者客户端参数及创建相应的生产者实例

（3）发送消息

我们直接上代码。

引入maven依赖

<dependency>
 		<groupId>org.apache.kafka</groupId>
 		<artifactId>kafka-clients</artifactId>
 		<version>3.3.0</version>
</dependency>

核心发送逻辑

public static void main(String[] args) {
    // 1. 创建 kafka 生产者的配置对象
    Properties properties = new Properties();
    // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    // key,value 序列化（必须）：key.serializer，value.serializer
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer");
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer");
    // 3. 创建 kafka 生产者对象
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
    // 4. 调用 send 方法,发送消息
    for (int i = 0; i < 5; i++) {
        kafkaProducer.send(new
                ProducerRecord<>("first", Integer.toString(i), "hello " + i));
    }
    // 5. 关闭资源
    kafkaProducer.close();
}

消息对象ProducerRecord

kafka发送时主要构造出ProducerRecord对象，包含发送的主题，partition，key，value等。

public class ProducerRecord<K, V> {

    private final String topic;
    private final Integer partition;
    private final Headers headers;
    private final K key;
    private final V value;
    private final Long timestamp;
}

三种发送模式

kafka提供了3种发送消息的模式，发后即忘，同步发送和异步发送，我们直接上代码。

发后即忘(

fire-and-forget

）

Future<RecordMetadata> send = producer.send(rcd);

同步发送（ sync ****）

只需在上面种发送方式的基础上，再调用一下 get()方法即可，该方法时阻塞的。

// 同步发送
 kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

带回调异步发送（ async ****）

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是 RecordMetadata 和Exception，如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

for (int i = 0; i < 5; i++) {
    kafkaProducer.send(new
            ProducerRecord<>("first", Integer.toString(i), "hello " + i), new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception == null) {
                // 没有异常,输出信息到控制台
                System.out.println(" 主题： " +
                        metadata.topic() + "->" + "分区：" + metadata.partition());
            } else {
                // 出现异常打印
                exception.printStackTrace();
            }
        }
    });
}

生产者发送核心机制

生产者分区机制

便于合理使用存储资源

Partition

Broker

提高并行度和吞吐量，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

那究竟生产者是按照什么样的策略发往到不同的分区呢？

DefaultPartitioner，具体逻辑如下：

按指定分区发送

kafka发送消息的时候构造消息对象ProducerRecord，可以传入指定的partition, 那么消息就会发送这个指定的分区。例如partition=0，所有数据写入分区0。

 // 发送消息到0号分区
kafkaProducer.send(new
        ProducerRecord<>("first", 0, Integer.toString(i), "hello " + i));

没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；

例如：key1的hash值=5，key2的hash值=6，topic的partition数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。

既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。

linger.ms设置的时间到，Kafka再随机一个分区进行使用（如果还是0会继续随机)。

自定义分区器

alvin，就发往 0 号分区，不包含 alvin，就发往 1 号分区。

Partitioner

/**
 * 1. 实现接口 Partitioner
 * 2. 实现 3 个方法:partition,close,configure
 * 3. 编写 partition 方法,返回分区号
 */
public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 获取消息
        String msgValue = value.toString();
        // 创建 partition
        int partition;
        // 判断消息是否包含 alvin
        if (msgValue.contains("alvin")){
            partition = 0;
        }else {
            partition = 1;
        }
        // 返回分区号
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

配置分区器

// 添加自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.alvin.kafka.producer.MyPartitioner");

// 发送消息 略~~

如何提高生产者吞吐量？

对比着前面kafka生产者的发送流程，kafka生产者提供的一些配置参数可以有助于提高生产者的吞吐量。

参数名称	描述
`buffer.memory`	`RecordAccumulator` 缓冲区总大小，默认 32m。适当增加该值，可以提高吞吐量。
`batch.size`	缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
`linger.ms`	如果数据迟迟未达到 `batch.size`，`sender`线程等待 `linger.time`之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 `5-100ms` 之间。
`compression.type`	指定消息的压缩方式，默认值为“`none` "，即默认情况下，消息不会被压缩。该参数还可以配置为 "`gzip`"，"`snappy`" 和 "`lz4`"。对消息进行压缩可以极大地减少网络传输、降低网络 I/O，从而提高整体的性能。

如何保证生产者消息的可靠性？

kafka 在 producer 里面提供了消息确认机制。我们可以通过配置来决定消息发送到对应分区的几个副本才算消息发送成功。可以在定义 producer 时通过 acks 参数指定。

acks=0

acks=1（默认值）

Leader收到数据后应答。

acks=-1或者all

Leader和ISR队列里面的所有节点收齐数据后应答。

ISR 概念：（同步副本）。每个分区的 leader 会维护一个 ISR 列表，ISR 列表里面就是 follower 副本的 Borker 编号，只有跟得上 Leader 的 follower 副本才能加入到 ISR 里面，这个是通过 replica.lag.time.max.ms =30000（默认值）参数配置的，只有 ISR 里的成员才有被选为 leader 的可能。

如果Leader收到数据，所有Follower都开始同步数据，但有一个Follower ，因为某种故障，迟迟不能与Leader进行同步，那这个问题怎么解决呢？

Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合(leader：0，isr:0,1,2)。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。

小结：数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2。

acks=0

acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
acks=-1或者all，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；

acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

如何保证消息只发送一次？

最多一次（At Most Once）

ack级别设置为0, 可以保证数据不重复，但是不能保证数据不丢失, 所以叫做最多一次。

至少一次（At Least Once）

ISR里应答的最小副本数量大于等于2可能会出现至少一次的消息。比如下图中在发送过程Leader节点宕机，消息就会重试，就有可能出现消息的重复。

At Least Once可以保证数据不丢失，但是不能保证数据不重复。

精确一次（Exactly Once）

幂等性和事务的特性实现。

精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2）。

enable.idempotence 设置为 true即可，在3.x版本中默认情况下也是true。具体实现原理如下：

producer

producer_id

partition

producer 每发送一条消息，会将<producer_id,分区>对应的“序列号”加 1。
broker 服务端端会为每一对<producer_id,分区>维护一个序列号，对于每收到的一条消息，会判断服务端的 SN_old 和接收到的消息中的 SN_new 进行对比：

如果 SN_OLD+1 = SN_NEW，正常情况
如果 SN_old+1>SN_new，说明是重复写入的数据，直接丢弃
如果 SN_old+1<SN_new，说明中间有数据尚未写入，或者是发生了乱序，或者是数据丢失，将抛出严重异常：OutOfOrderSequenceException。

如何保证生产者消息的顺序？

kafka在1.x版本之前保证数据单分区有序，条件如下：

max.in.flight.requests.per.connection=1

kafka在1.x及以后版本保证数据单分区有序，条件如下：

max.in.flight.requests.per.connection

开启幂等性，max.in.flight.requests.per.connection需要设置小于等于5。

因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。

总结

欢迎关注个人公众号【JAVA旭阳】交流学习

编程笔记 » kafka生产者你不得不知的那些事儿