《深入理解kafka-核心设计与实践原理》第二章：生产者

第二章：生产者

2.1 api使用

2.1.1 生产者：KafkaProducer

2.1.2 消息体：ProducerRecord

2.1.3 序列化器：Serializer

2.1.4 分区器：Partitioner

2.1.5 拦截器：Interceptor

2.2 生产者客户端的原理

2.2.1 整体架构

2.2.2 元数据的更新

2.3 重要的生产者参数

2.3.1 acks

2.3.2 max.request.size

2.3.3 retries和retry.backoff.ms

2.3.4 其他参数

第二章：生产者

2.1 api使用

2.1.1 生产者：KafkaProducer

• ①KafkaProducer是线程安全的，可以在多个线程中共享单个 KafkaProducer 实例
• ②kafkaProducer中的必要参数
  • bootstrap.servers：指定生产者客户端连接 Kafka 集群所需的 broker 地址清单，具体的内容格式为 host1:port1, host2:port2
    • 注意这里并非需要所有的 broker 地址，因为生产者会从给定的 broker 里查找到其他 broker 的信息 。不过建议至少要设置 两个以上的 broker 地址信息，当其中任意 一个岩机时，生产者仍然可以连接到 Kafka 集群上
  • key.serializer 和 value.serializer：指定序列化器，这里需写序列化器的全限类名
    • 生产者客户端在发往 broker之前需要将消息中对应的 key 和 value 做相应的序列化操作来转换成字节数组
• ③构造器

  • KafkaProducer<String, String> producer= new KafkaProducer<>(props)
    KafkaProducer<String, String> producer= new KafkaProducer<>(props, new StringSerializer() , new StringSerializer())
    

• ④消息发送
  • 发送的对象：ProducerRecord（下文会提到）
• ⑤消息发送api

  • public Future<RecordMetadata> send(ProducerRecord<K, V> record)
    public Future<RecordMetadata> send(ProducerRecord<K , V> record , Callback callback)
    

• ⑥发送消息的三种模式
  • 异步(async)：在 send()方法里指定一个 Callback 的回调函数
    • 对于同一个分区而言，如果消息 record1 于 record2 之前先发送，那么KafkaProducer就可以保证对应的 callback1 在 callback2 之前调用，也就是说，回调函数的调用也可以保证分区有序
  • 同步(sync)：可靠性高，但性能差。要么消息被发送成功，要么发生异常
    • 一般会发生两种类型的异常：可重试的异常和不可重试的异常，对于可重试的异常，如果配置了 retries 参数，那么只要在规定的重试次数内自行恢复了，就不会抛出异常。retries参数的默认值为0

    • Future<RecordMetadata> future = producer.send(record); 
      RecordMetadata metadata = future.get();
      

  • 发后即忘(fire-and-forget)：它只管往 Kafka 中发送消息而并不关心消息是否正确到达，消息可能会丢失。性能最高，可靠性也最差
• ⑦RecordMetadata：RecordMetadata对象里包含了消息的一些元数据信息，比如当前消息的主题、分区号、分区中的偏移量(offset)、时间戳等

2.1.2 消息体：ProducerRecord

• 消息对象，有以下几种构造方法

  • public ProducerRecord(String topic, Integer partition, Long timestamp,K key, V value, Iterable<Header> headers)
     
    public ProducerRecord(String topic, Integer partition， Long timestamp,K key , V value)
     
    public ProducerRecord(String topic , Integer partition， k key, V value,Iterable<Header> headers)
     
    public ProducerRecord(String topic, Integer partition, K key, V value)public ProducerRecord(String topic, K key, V value)
     
    public ProducerRecord(String topic, V value)
    

2.1.3 序列化器：Serializer

• 对象转换成字节数组才能通过网络发送给 Kafka，自带的序列化器在org.apache.kafka.common.serialization下，都实现了org.apache.kafka.common.serialization.Serializer接口

  • public void configure (Map<String , ?> configs , boolean isKey)	//配置当前类
    public byte[] serialize(String topic , T data)	//执行序列化操作
    public void close()	//关闭当前的序列化器，
    

• 一般情况下 encoding 的值就为默认的"UTF-8"

2.1.4 分区器：Partitioner

• 消息在通过send()方法发往broker的过程中， 需要经过拦截器(Interceptor)(非必须)、序列化器(Serializer)(必须)和分区器(Partitioner)(非必须)之后才能被真正地发往broker
• 如果消息ProducerRecord中没有指定partition字段，那么就需要依赖分区器。如果有指定partition，那么就无需分区器，因为partition代表的就是所要发往的分区号
• Kafka中提供的默认分区器是org.apache.kafka.clients.producer.intemals.DefaultPartitioner(对key进行哈希)，它实现了org.apache.kafka.clients.producer.Partitioner接口

  • //计算分区号，返回值为int类型
    public int partition(String topic , Object key, byte[] keyBytes , Object value , byte[] valueBytes , Cluster cluster);
    //关闭分区器时回收资源的方法
    public void close();
    

• 默认分区器：DefaultPartitioner
  • 如果 key != null：通过计算哈希值来获得分区号，拥有相同key的消息会被写入同一个分区
  • 如果 key == null，那么消息将会以轮询的方式发往topic内的各个可用分区
  • 问题：一旦增加/减少了partition，那么就难以保证key与原先partition之间的映射关系了
  • 自定义分区器：实现Partitioner类

2.1.5 拦截器：Interceptor

• Kafka一共有两种拦截器 : 生产者拦截器和消费者拦截器，本节讲述生产者相关内容
• 生产者拦截器：生产者拦截器既可以用来在消息发送前做一些准备工作
  • 例：按照某个规则过滤不符合要求的消息、修改消息的内容等
  • 例：在发送回调逻辑前做一些定制化的需求，比如统计类工作
• 自定义拦截器，需实现org.apache.kafka.clients.producer.Producerlnterceptor接口。然后在配置参数中添加这个拦截器。

  • //KafkaProducer在将消息序列化和计算分区之前会调用生产者拦截器的onSend()方法来对消息进行相应的定制化操作
    public ProducerRecord<K, V> onSend (ProducerRecord<K, V> record);
     
    //在消息被应答之前或消息发送失败时调用生产者拦截器的onAcknowledgement()方法，该方法会优先于用户设定的Callback之前执行
    public void onAcknowledgement(RecordMetadata metadata, Exception exception ); 
     
    //主要用于在关闭拦截器时执行一些资源的清理工作
    public void close() ;
    

• KafkaProducer中可以指定多个拦截器以形成拦截链。拦截链会按照interceptor.classes参数配置的拦截器的顺序来一一执行

  • //生产者的interceptor.classes配置
    properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerinterceptorV1.class.getName() + ”,” + ProducerinterceptorV2.class.getName());
    

2.2 生产者客户端的原理

2.2.1 整体架构

线程：整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和Sender线程(发送线程)

• 主线程：在主线程中由 KafkaProducer 创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器(RecordAccumulator，也称为消息收集器)中
• Sender线程：负责从 RecordAccumulator 中获取消息并将其发送到Kafka服务端中

消息累加器(RecordAccumulator)

• 主要用来缓存消息以便 Sender 线程可以批量发送，进而减少网络传输 的资源消耗以提升性能。缓存大小由buffer.memory配置，默认32MB
• 如果【主线程生产者发送消息的速度】超过【Sender线程发送到服务器的速度】，则会导致生产者空间不足，这个时候KafkaProducer的send()方法调用要么被阻塞，要么抛出异常。阻塞的最长等待时间max.block.ms配置，默认60s
• RecordAccumulator内部为每个分区维护了一个双端队列(Deque<ProducerBatch>)，主线程写入时从尾部写，Sender线程读时从头部读
  • ProducerBatch：一个消息批次，包含多个ProducerRecord，用于提升传输效率。大小由batch.size参数指定，默认16KB
• BufferPool：频繁的创建和释放内存是比较耗费资源的，在RecordAccumulator的内部维护了一个BufferPool，它主要用来实现 ByteBuffer 的复用，以实现缓存的高效利用
  • 可以适当地调大batch.size参数以便多缓存一些消息
• ProducerRecord进入RecordAccumulator后的整个流程
  • ①先寻找与消息分区所对应的双端队列(如果没有则新建)
  • ②从这个双端队列的尾部获取一个ProducerBatch(如果没有则新建)
  • ③查看ProducerBatch中是否还可以写入这个ProducerRecord。如果可以则写入，如果不可以则需要创建一个新的ProducerBatch
  • ④在新建ProducerBatch时评估这条消息的大小是否超过 batch.size 参数的大小。如果不超过，那么就以 batch.size 参数的大小来创建ProducerBatch，这样在使用完这段内存区域之后，可以通过BufferPool的管理来进行复用；如果超过，那么就以评估的大小来创建ProducerBatch，这段内存区域不会被复用

Sender线程

• 从RecordAccumulator中获取消息后，会将<分区, Deque< ProducerBatch>>转变成<Node, List<ProducerBatch>>，最后转变成<Node, Request>，发送至对应的Node节点(broker节点)
• 请求在从Sender线程发往Kafka之前还会保存到InFlightRequests中，InFlightRequests保存对象的具体形式为Map<Nodeld, Deque<Request>>，它的主要作用是缓存了已经发出去但还没有收到响应的请求

2.2.2 元数据的更新

• 元数据的概念：元数据是指kafka集群的元数据。如集群中有哪些主题，这些主题有哪些分区，每个分区的leader副本分配在哪个节点上，follower副本分配在哪些节点上，哪些副本在AR、ISR 等集合中，集群中有哪些节点，控制器节点又是哪一个等信息
• 如何更新元数据
  • 什么是leastLoadedNode：sender线程通过InFlightRequests负责更新，InFlightRequests可以获得leastLoadedNode(所有 Node 中负载最小的那一个)，这里的负载最小是通过每个Node在InFlightRequests中还未确认的请求决定的，下图中Node1就是leastLoadedNode
  • 何时更新元数据：当客户端中没有需要使用的元数据信息时，比如没有指定的主题信息，或者超过metadata.max.age.rns时间(默认5分钟)没有更新元数据，都会引起元数据的更新操作
  • 怎么更新元数据：当需要更新元数据时，会先挑选出leastLoadedNode，然后向这个Node发送 MetadataRequest请求来获取具体的元数据信息。这个更新操作是由 Sender 线程发起的，在创建完 MetadataRequest之后同样会存入InF!ightRequests
    • 

2.3 重要的生产者参数

本小节主要讲生产者客户端的参数

2.3.1 acks

生产者客户端中一个非常重要的参数，可靠性和吞吐量之前的权衡策略，默认值acks = 1

• acks = 0：生产者发送过来的数据，不需要等数据落盘应答
• acks = 1：默认值。生产者发送消息之后，生产者发送过来的数据，需要等Leader成功写入数据后应答(不关心follower是否写入成功)
• acks = -1 或 acks = all：生产者发送过来的数据，Leader和ISR队列里面的所有结点收到数据后应答

分析

• acks = 0：leader收到数据后，未落盘前宕机，则数据丢失
  • 可靠性差，但能达到最大吞吐量。基本不用该模式。
• acks = 1：leader收到数据并应答后，还未来得及与follower同步便宕机，则数据丢失
  • 可靠性中等，比ack=0强，但是仍不完全可靠，是可靠性和吞吐量折中方案。适用于传输日志等
• acks = -1：leader收到数据并落盘，follower向leader拉取数据落盘后响应leader，leader收到全部响应后应答生产者
  • 分析可能存在的问题：follower挂了，没有应答leader，leader是否会无限等待
    • 结论：不会。leader会维护一个动态的in-sync replica set(ISR)，意为和leader保持同步的follower+leader集合（eg.leader=0, ISR={0,1,2}）。如果follower长时间未向leader发送通信请求或同步数据，则该follower被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s
  • 可靠性有保障(除非所有follower都挂了)，但效率差。适用于数据必须可靠的场景
• 其他
  • 除了acks参数外，还有其他几个配置项可以与之配合使用来进一步保证消息的可靠性
    • min.insync.replicas：设置在认为消息写入是成功的之前，必须有多少个副本已经收到该消息
    • retries和retry.backoff.ms：设置生产者在发送失败时重试的次数和重试间隔
    • max.in.flight.requests.per.connection：设置在收到服务器响应之前，生产者端允许发送的未确认请求的最大数量

2.3.2 max.request.size

生产者客户端能发送的消息的最大值，默认1MB

• 不建议盲目增大这个值，因为这个参数还涉及一些其他参数的联动，比如broker端的 message.max.bytes 参数(broker端能接受的消息最大值)

2.3.3 retries和retry.backoff.ms

• retries：配置生产者重试的次数，默认值为 0，即在发生异常的时候不进行任何重试动作
  • 可适用于一些临时性的异常，比如网络抖动、leader副本的选举等，这种异常往往是可以自行恢复的
  • 有些异常不适用：如消息太大，再重试也无用
• retry.backoff.ms：两次重试之间的时间间隔，默认100ms

2.3.4 其他参数

• compression.type：消息的压缩方式，默认为none，不压缩。可以选择"gzip"，"snappy" 和 "lz4"，时间换空间。如果对时延有一定的要求，则不推荐对消息进行压缩
• connections.max.idle.ms：指定在多久之后关闭限制的连接，默认值是 540000(ms)，即9分钟
• linger.ms：生产者发送ProducerBatch之前等待更多消息(ProducerRecord)加入ProducerBatch的时间。
  • 默认值为0，即不等待直接发送。可适当调大
  • 生产者客户端会在ProducerBatch被填满或等待时间超过linger.ms值时发送出去
  • 增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞 吐量
• request.timeout.ms：这个参数用来配置Producer等待请求响应的最长时间，默认值为30000(ms)。超时后可以选择进行重试