由浅入深！全面了解Kafka 生产者解析，赶紧拿下

﹏ヽ暗。殇╰゛Y 2023-09-28 10:26 83阅读 0赞

## 一、消息发送 ##

## 1.1 数据生产流程 ##

数据生产流程图解：

![cadb70bb947680189bf7be00f1c50c82.png][]

1.  Producer创建时，会创建⼀个Sender线程并设置为守护线程
2.  ⽣产消息时，内部其实是异步流程；⽣产的消息先经过**拦截器->序列化器->分区器**，然后将消息缓存在缓冲区（该缓冲区也是在Producer创建时创建）
3.  批次发送的条件为：缓冲区数据⼤⼩达到 batch.size 或者 linger.ms 达到上限，哪个先达到就算哪个
4.  批次发送后，发往指定分区，然后落盘到 broker；如果⽣产者配置了retrires参数⼤于0并且失败原因允许重试，那么客户端内部会对该消息进⾏重试
5.  落盘到broker成功，返回⽣产元数据给⽣产者
6.  元数据返回有两种⽅式：⼀种是通过阻塞直接返回，另⼀种是通过回调返回

## 1.2 必要的参数配置 ##

先来看看我们一般在程序中是怎么配置的：

![cd9a78ae0741b5fe9bf88e899c26aa58.png][]

最常用的配置项：

<table> 
 <tbody> 
 <tr> 
 <td> 属性 </td> 
 <td> 说明 </td> 
 <td> 重要性 </td> 
 </tr> 
 <tr> 
 <td> bootstrap.servers </td> 
 <td> ⽣产者客户端与broker集群建⽴初始连接需要的broker地址列表，由该初始连接发现Kafka集群中其他的所有broker。该地址列表不需要写全部的Kafka集群中broker的地址，但也不要写⼀个，以防该节点宕机的时候不可⽤。形式为：host1:port1,host2:port2,.... </td> 
 <td> high </td> 
 </tr> 
 <tr> 
 <td> key.serializer </td> 
 <td> 实现了接⼝ org.apache.kafka.common.serialization.Serializer的key序列化类。 </td> 
 <td> high </td> 
 </tr> 
 <tr> 
 <td> value.serializer </td> 
 <td> 实现了接⼝ org.apache.kafka.common.serialization.Serializer的value序列化类。 </td> 
 <td> high </td> 
 </tr> 
 <tr> 
 <td> acks </td> 
 <td> 该选项控制着已发送消息的持久性。 acks=0：⽣产者不等待broker的任何消息确认。只要将消息放到了socket的缓冲区，就认为消息已发送。不能保证服务器是否收到该消息，retries设置也不起作⽤，因为客户端不关⼼消息是否发送失败。客户端收到的消息偏移量永远是-1。 acks=1：leader将记录写到它本地⽇志，就响应客户端确认消息，⽽不等待follower副本的确认。如果leader确认了消息就宕机，则可能会丢失消息，因为follower副本可能还没来得及同步该消息。 acks=all：leader等待所有同步的副本确认该消息。保证了只要有⼀个同步副本存在，消息就不会丢失。这是最强的可⽤性保证。等价于acks=-1。默认值为1，字符串。可选值：[all, -1, 0, 1] </td> 
 <td> high </td> 
 </tr> 
 <tr> 
 <td> compression.type </td> 
 <td> ⽣产者⽣成数据的压缩格式。默认是none（没有压缩）。允许的值：none，gzip，snappy和lz4。压缩是对整个消息批次来讲的。消息批的效率也影响压缩的⽐例。消息批越⼤，压缩效率越好。字符串类型的值。默认是none。 </td> 
 <td> high </td> 
 </tr> 
 <tr> 
 <td> retries </td> 
 <td> 设置该属性为⼀个⼤于1的值，将在消息发送失败的时候重新发送消息。该重试与客户端收到异常重新发送并⽆⼆⾄。允许重试但是不设置 max.in.flight.requests.per.connection为 1，存在消息乱序的可能，因为如果两个批次发送到同⼀个分区，第⼀个失败了重试，第⼆个成功了，则第⼀个消息批在第⼆个消息批后。int类型的值，默认：0，可选值：[0,...,2147483647] </td> 
 <td> high </td> 
 </tr> 
 </tbody> 
</table>

## 1.3 拦截器 ##

## 1.3.1 拦截器介绍 ##

Producer 的拦截器（Interceptor）和 Consumer 的 Interceptor 主要⽤于实现Client端的定制化控制逻辑。  
对于Producer⽽⾔，Interceptor使得⽤户在消息发送前以及Producer回调逻辑前有机会对消息做⼀些定制化需求，⽐如修改消息等。同时，Producer允许⽤户指定多个Interceptor按序作⽤于同⼀条消息从⽽形成⼀个拦截链(Interceptor Chain)。Intercetpor 的实现接⼝是  
org.apache.kafka.clients.producer.ProducerInterceptor，其定义的⽅法包括：

*  onSend(ProducerRecord)：该⽅法封装进KafkaProducer.send⽅法中，即运⾏在⽤户主线程中。Producer确保在消息被序列化以计算分区前调⽤该⽅法。⽤户可以在该⽅法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响⽬标分区的计算。
 *  onAcknowledgement(RecordMetadata, Exception)：该⽅法会在消息被应答之前或消息发送失败时调⽤，并且通常都是在Producer回调逻辑触发之前。onAcknowledgement运⾏在Producer的IO线程中，因此不要在该⽅法中放⼊很重的逻辑，否则会拖慢Producer的消息发送效率。
 *  close：关闭Interceptor，主要⽤于执⾏⼀些资源清理⼯作。

如前所述，Interceptor可能被运⾏在多个线程中，因此在具体实现时⽤户需要**⾃⾏确保线程安全**。另外倘若指定了多个Interceptor，则Producer将按照指定顺序调⽤它们，并仅仅是捕获每个Interceptor可能抛出的异常记录到错误⽇志中⽽⾮在向上传递。这在使⽤过程中要特别留意。

## 1.3.2 自定义拦截器 ##

自定义拦截器步骤：

1.  实现ProducerInterceptor接⼝
2.  在KafkaProducer的设置中设置⾃定义的拦截器

**自定义拦截器 1**：

public class InterceptorOne<Key, Value> implements ProducerInterceptor<Key, Value> { private static final Logger LOGGER = LoggerFactory.getLogger(InterceptorOne.class); @

[cadb70bb947680189bf7be00f1c50c82.png]: https://img-blog.csdnimg.cn/img_convert/cadb70bb947680189bf7be00f1c50c82.png
[cd9a78ae0741b5fe9bf88e899c26aa58.png]: https://img-blog.csdnimg.cn/img_convert/cd9a78ae0741b5fe9bf88e899c26aa58.png