大数据-消息队列-Kafka(五)：Producer(生产者)【发送消息采用的是异步发送】【两个线程：main线程和Sender线程】【线程共享变量：双端队列RecordAccumulator】

逃离我推掉我的手 2023-09-25 21:18 7阅读 0赞

## **一、生产者消息发送流程** ##

### **1、发送原理** ###

Kafka的Producer发送消息采用的是**异步发送**的方式。

在消息发送的过程中，涉及到了**两个线程****:***main**线程*和*Sender**线程*，以及一个*线程共享变量*:RecordAccumulator。

*  ①main线程中创建了一个双端队列RecordAccumulator，将消息发送给RecordAccumulator。
 *  ②Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

### **2、生产者重要参数列表** ###

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

## **二、异步发送API** ##

### **1、普通异步发送** ###

#### 1.1 需求 ####

创建Kafka生产者，采用异步的方式发送到Kafka broker

#### 1.2 异步发送流程 ####

![b989496b8c3d4cb497fb161b788071c0.gif][]

![ba781591412a482591e03e7ec24dfd1e.png][]

失败后重试的次数是int的最大值；

#### 1.3 代码编写 ####

1）创建工程kafka-demo

2）导入依赖

<dependencies>
 <dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 <version>3.0.0</version>
 </dependency>
 </dependencies>

3）创建包名：com.wyr.kafka.producer

4）编写代码：不带回调函数的API

package com.atguigu.kafka.producer;
    
    imp

[b989496b8c3d4cb497fb161b788071c0.gif]: https://img-blog.csdnimg.cn/b989496b8c3d4cb497fb161b788071c0.gif
[ba781591412a482591e03e7ec24dfd1e.png]: https://img-blog.csdnimg.cn/ba781591412a482591e03e7ec24dfd1e.png