大数据-消息队列-Kafka(七)：Consumer(消费者)

客官°小女子只卖身不卖艺 2023-10-08 23:49 106阅读 0赞

### 5.1 Kafka消费方式 ###

![7316a1c1e2774257b6bbfb15d349f624.png][]

*Kafka**的consumer**采用pull**（拉）模式从broker**中读取数据。*

<table> 
 <tbody> 
 <tr> 
 <td style="background-color:#ccffcc;border-color:#000000;width:84.8pt;"> push（推）模式 </td> 
 <td style="vertical-align:top;width:437.4pt;"> 很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。 </td> 
 </tr> 
 <tr> 
 <td style="background-color:#ccffcc;width:84.8pt;"> pull模式 </td> 
 <td style="vertical-align:top;width:437.4pt;"> 可以根据consumer的消费能力以适当的速率消费消息。不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。 </td> 
 </tr> 
 </tbody> 
</table>

### 5.2 Kakfa消费者工作流程 ###

#### 5.2.1 消费者总体工作流程 ####

![f5b25bd5c13d4104b25515c466c0084d.png][]

#### 5.2.2 消费者组原理 ####

![2691ed72bf77451ea773b186d0489fce.png][]

![c0d22e906ccd4684a96722fa01f57519.png][]

![f66de4700b194f6785be76e106177666.png][]

![ef1c8f92fb10469ca2fbe1fb7d89f2c8.png][]

#### 5.2.3 消费者重要参数 ####

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

### 5.3 消费者API ###

#### 5.3.1 独立消费者案例（订阅主题） ####

![ab34cd4000154824913bad43cad3a362.png][]

[7316a1c1e2774257b6bbfb15d349f624.png]: https://img-blog.csdnimg.cn/7316a1c1e2774257b6bbfb15d349f624.png
[f5b25bd5c13d4104b25515c466c0084d.png]: https://img-blog.csdnimg.cn/f5b25bd5c13d4104b25515c466c0084d.png
[2691ed72bf77451ea773b186d0489fce.png]: https://img-blog.csdnimg.cn/2691ed72bf77451ea773b186d0489fce.png
[c0d22e906ccd4684a96722fa01f57519.png]: https://img-blog.csdnimg.cn/c0d22e906ccd4684a96722fa01f57519.png
[f66de4700b194f6785be76e106177666.png]: https://img-blog.csdnimg.cn/f66de4700b194f6785be76e106177666.png
[ef1c8f92fb10469ca2fbe1fb7d89f2c8.png]: https://img-blog.csdnimg.cn/ef1c8f92fb10469ca2fbe1fb7d89f2c8.png
[ab34cd4000154824913bad43cad3a362.png]: https://img-blog.csdnimg.cn/ab34cd4000154824913bad43cad3a362.png