Kafka：消费者调优和参数配置

清疚 2024-03-24 15:14 116阅读 0赞

### 消费者组初始化流程 ###

![b341942fdecc4fa2873dee6a8bbc7139.png][]

### 消费者组详细消费流程 ###

![87e8e2c236c44547b6e8a6e897b27304.png][]

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>向 Kafka 集群建立初始连接用到的 host/port 列表。</td> 
  </tr> 
  <tr> 
   <td>key.deserializer 和 value.deserializer</td> 
   <td>指定接收消息的 key 和 value 的反序列化类型。一定要写全 类名。</td> 
  </tr> 
  <tr> 
   <td>group.id</td> 
   <td>标记消费者所属的消费者组。</td> 
  </tr> 
  <tr> 
   <td>enable.auto.commit</td> 
   <td><span style="color:#fe2c24;">默认值为 true</span>，消费者会自动周期性地向服务器提交偏移量。</td> 
  </tr> 
  <tr> 
   <td>auto.commit.interval.ms</td> 
   <td>如果设置了 enable.auto.commit 的值为 true， 则该值定义了 消费者偏移量向 Kafka 提交的频率，<span style="color:#fe2c24;">默认 5s</span>。</td> 
  </tr> 
  <tr> 
   <td>auto.offset.reset</td> 
   <td>当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在 （如，数据被删除了），该如何处理？ earliest：自动重置偏 移量到最早的偏移量。 <span style="color:#fe2c24;">latest：默认，自动重置偏移量为最新 的偏移量</span>。 none：如果消费组原来的（previous）偏移量不 存在，则向消费者抛异常。 anything：向消费者抛异常。</td> 
  </tr> 
  <tr> 
   <td>offsets.topic.num.partitions</td> 
   <td>__consumer_offsets 的分区数，<span style="color:#fe2c24;">默认是 50 个分区。不建议修改</span>。</td> 
  </tr> 
  <tr> 
   <td>heartbeat.interval.ms</td> 
   <td>Kafka 消费者和 coordinator 之间的心跳时间，<span style="color:#fe2c24;">默认 3s</span>。 该条目的值必须小于 session.timeout.ms ，也不应该高于 session.timeout.ms 的 1/3。<span style="color:#fe2c24;">不建议修改</span>。</td> 
  </tr> 
  <tr> 
   <td>session.timeout.ms</td> 
   <td>Kafka 消费者和 coordinator 之间连接超时时间，<span style="color:#fe2c24;">默认 45s</span>。超 过该值，该消费者被移除，消费者组执行再平衡。</td> 
  </tr> 
  <tr> 
   <td>max.poll.interval.ms</td> 
   <td>消费者处理消息的最大时长，<span style="color:#fe2c24;">默认是 5 分钟</span>。超过该值，该 消费者被移除，消费者组执行再平衡。</td> 
  </tr> 
  <tr> 
   <td>fetch.min.bytes</td> 
   <td><span style="color:#fe2c24;">默认 1 个字节。消费</span>者获取服务器端一批消息最小的字节数。</td> 
  </tr> 
  <tr> 
   <td>fetch.max.wait.ms</td> 
   <td><span style="color:#fe2c24;">默认 500ms。如果没有从服务器端获取到一批数据的最小字节数</span>。该时间到，仍然会返回数据。</td> 
  </tr> 
  <tr> 
   <td>fetch.max.bytes</td> 
   <td>默认<span style="color:#fe2c24;"> Default: 52428800（50 m）。消费者</span>获取服务器端一批 消息最大的字节数。如果服务器端一批次的数据大于该值 （50m）仍然可以拉取回来这批数据，因此，这不是一个绝 对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。</td> 
  </tr> 
  <tr> 
   <td>max.poll.records</td> 
   <td>一次 poll 拉取数据返回消息的最大条数，<span style="color:#fe2c24;">默认是 500 条</span>。</td> 
  </tr> 
 </tbody> 
</table>

### **消费者再平衡** ###

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>参数名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>heartbeat.interval.ms</td> 
   <td>Kafka 消费者和 coordinator 之间的心跳时间，<span style="color:#fe2c24;">默认 3s</span>。 该条目的值必须小于 session.timeout.ms，也不应该高于 session.timeout.ms 的 1/3。</td> 
  </tr> 
  <tr> 
   <td>session.timeout.ms</td> 
   <td>Kafka 消费者和 coordinator 之间连接超时时间，<span style="color:#fe2c24;">默认 45s</span>。 超过该值，该消费者被移除，消费者组执行再平衡。</td> 
  </tr> 
  <tr> 
   <td>max.poll.interval.ms</td> 
   <td>消费者处理消息的最大时长，<span style="color:#fe2c24;">默认是 5 分钟</span>。超过该值，该 消费者被移除，消费者组执行再平衡。</td> 
  </tr> 
  <tr> 
   <td>partition.assignment.strategy</td> 
   <td>消 费 者 分 区 分 配 策 略 ， 默 认 策 略 是 Range + CooperativeSticky。Kafka 可以同时使用多个分区分配策略。 可以选择的策略包括：<span style="color:#fe2c24;">Range、RoundRobin、Sticky、 CooperativeSticky</span></td> 
  </tr> 
 </tbody> 
</table>

[b341942fdecc4fa2873dee6a8bbc7139.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/6e373f525db741dc9e0dc23d6732171c.png
[87e8e2c236c44547b6e8a6e897b27304.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/24/b18cbc96a57f4f2f83a65c99210d4f05.png