Kafka：生产者调优和参数配置

傷城~ 2024-03-24 14:06 59阅读 0赞

### 发送流程 ###

![f2a8bc74f2ac47d799449834671c499e.png][]

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>生 产 者 连 接 集 群 所 需 的 broker 地 址 清 单 。 例 如 hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需 要所有的 broker 地址，因为生产者从给定的 broker 里查 找到其他 broker 信息。</td> 
  </tr> 
  <tr> 
   <td>key.serializer 和 value.serializer</td> 
   <td>指定发送消息的 key 和 value 的序列化类型。一定要写 全类名。</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>RecordAccumulator 缓冲区总大小，默认 32m。</td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>缓冲区一批数据最大值，默认 16k。适当增加该值，可 以提高吞吐量，但是如果该值设置太大，会导致数据传 输延迟增加。</td> 
  </tr> 
  <tr> 
   <td>linger.ms</td> 
   <td>如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。单位 ms，默认值是 0ms，表示没有 延迟。生产环境建议该值大小为 5-100ms 之间。</td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td> <p>0：生产者发送过来的数据，不需要等数据落盘应答。</p> <p>1：生产者发送过来的数据，Leader 收到数据后应答。</p> <p>-1（all）：生产者发送过来的数据，Leader+和 isr 队列 里面的所有节点收齐数据后应答。默认值是-1，-1 和 all 是等价的。</p> </td> 
  </tr> 
  <tr> 
   <td>max.in.flight.requests.per.connection</td> 
   <td>允许最多没有返回 ack 的次数，默认为 5，开启幂等性 要保证该值是 1-5 的数字。</td> 
  </tr> 
  <tr> 
   <td>retries</td> 
   <td>当消息发送出现错误的时候，系统会重发消息。retries 表 示重试次数。默认是 int 最大值，2147483647。 如果设置了重试，还想保证消息的有序性，需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送成 功了。</td> 
  </tr> 
  <tr> 
   <td>retry.backoff.ms</td> 
   <td>两次重试之间的时间间隔，默认是 100ms。</td> 
  </tr> 
  <tr> 
   <td>enable.idempotence</td> 
   <td>是否开启幂等性，默认 true，开启幂等性。</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td> <p>生产者发送的所有数据的压缩方式。默认是 none，也就 是不压缩。</p> <p><span style="color:#fe2c24;">支持压缩类型：none、gzip、snappy、lz4 和 zstd。</span></p> </td> 
  </tr> 
 </tbody> 
</table>

### 生产者如何提高吞吐量 ###

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>RecordAccumulator 缓冲区总大小，默认 32m。</td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>缓冲区一批数据最大值，默认 16k。适当增加该值，可 以提高吞吐量，但是如果该值设置太大，会导致数据传 输延迟增加。</td> 
  </tr> 
  <tr> 
   <td>linger.ms</td> 
   <td>如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。单位 ms，默认值是 0ms，表示没有 延迟。生产环境建议该值大小为 5-100ms 之间。</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td> <p>生产者发送的所有数据的压缩方式。默认是 none，也就 是不压缩。</p> <p><span style="color:#fe2c24;">支持压缩类型：none、gzip、snappy、lz4 和 zstd。</span></p> </td> 
  </tr> 
 </tbody> 
</table>

### 数据可靠性 ###

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td> <p>0：生产者发送过来的数据，不需要等数据落盘应答。</p> <p>1：生产者发送过来的数据，Leader 收到数据后应答。</p> <p>-1（all）：生产者发送过来的数据，Leader+和 isr 队列 里面的所有节点收齐数据后应答。<span style="color:#fe2c24;">默认值是-1，-1 和 all 是等价的。</span></p> </td> 
  </tr> 
 </tbody> 
</table>

至少一次（At Least Once）= ACK 级别设置为-1 + 分区副本大于等于 2 + ISR 里应 答的最小副本数量大于等于 2

### 数据去重 ###

1）配置参数

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>enable.idempotence</td> 
   <td>是否开启幂等性，<span style="color:#fe2c24;">默认 true</span>，表示开启幂等性。</td> 
  </tr> 
 </tbody> 
</table>

2）Kafka 的事务一共有如下 5 个 API

// 1 初始化事务
    void initTransactions();
    
    // 2 开启事务
    void beginTransaction() throws ProducerFencedException;
    
    // 3 在事务内提交已经消费的偏移量（主要用于消费者）
    void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
    
    // 4 提交事务
    void commitTransaction() throws ProducerFencedException;
    
    // 5 放弃事务（类似于回滚事务的操作）
    void abortTransaction() throws ProducerFencedException;

### 数据有序 ###

单分区内，有序（有条件的，不能乱序）；多分区，分区与分区间无序；

### 数据乱序 ###

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>enable.idempotence</td> 
   <td>是否开启幂等性，默认 true，表示开启幂等性。</td> 
  </tr> 
  <tr> 
   <td>max.in.flight.requests.per.connection</td> 
   <td>允许最多没有返回 ack 的次数，默认为 5，开启幂等性 要保证该值是 1-5 的数字。</td> 
  </tr> 
 </tbody> 
</table>

[f2a8bc74f2ac47d799449834671c499e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/b45d9a2e4cf94bb682aa80a71df0030f.png