Kafka快速入门（生产者）同步异步发送、分区、消息精确一次发送、幂等性、事务

﹏ヽ暗。殇╰゛Y 2023-09-29 12:54 15阅读 0赞

## **Kafka** 生产者 ##

### 1. **生产者消息发送流程** ###

#### **1.1** **发送原理** ####

在消息发送的过程中，涉及到了两个线程——main **线程和** **Sender** **线程**。在 main 线程中创建了**一个双端队列** **RecordAccumulator**。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

![在这里插入图片描述][04aa5562bb664029890ea98a5fbff37b.png]

#### 1.2 **生产者重要参数列表** ####

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

### 2.**异步发送** **API** ###

#### 2.1 **普通异步发送** ####

1）需求：创建 Kafka 生产者，采用异步的方式发送到 Kafka Broker

**异步发送流程**

![在这里插入图片描述][7cb451f82a3249a0aa5c732bc41a87a8.png]

2）代码编写

（1）创建工程 kafka

（2）导入依赖

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

（3）创建包名：com.atguigu.kafka.producer

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

package com.atguigu.kafka.producer;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import java.util.Properties;
    
    public class CustomProducer {
        
         public static void main(String[] args) throws 
            InterruptedException {
        
             // 1. 创建 kafka 生产者的配置对象
             Properties properties = new Properties();
             // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
             properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
    
             // key,value 序列化（必须）：key.serializer，value.serializer
             properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
             // 3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new 
            KafkaProducer<String, String>(properties);
             // 4. 调用 send 方法,发送消息
             for (int i = 0; i < 5; i++) {
        
                 kafkaProducer.send(new 
                 ProducerRecord<>("first","yy " + i));
             }
             // 5. 关闭资源
             kafkaProducer.close();
      }
    }

测试：

①在 hadoop102 上开启 Kafka 消费者。

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

②在 IDEA 中执行代码，观察 hadoop102 控制台中是否接收到消息。

#### 2.2 带回调函数的异步发送 ####

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

// 1. 创建 kafka 生产者的配置对象
     Properties properties = new Properties();
     // 2. 给 kafka 配置对象添加配置信息
     properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
    "hadoop102:9092");
     // key,value 序列化（必须）：key.serializer，value.serializer
     properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     
     properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     // 3. 创建 kafka 生产者对象
     KafkaProducer<String, String> kafkaProducer = new 
     KafkaProducer<String, String>(properties);
     // 4. 调用 send 方法,发送消息
     for (int i = 0; i < 5; i++) {
        
         // 添加回调
         kafkaProducer.send(new ProducerRecord<>("first", "atguigu " + i), new Callback() {
        
                 // 该方法在 Producer 收到 ack 时调用，为异步调用
             @Override
             public void onCompletion(RecordMetadata metadata, 
                Exception exception) {
        
                 if (exception == null) {
        
                   // 没有异常,输出信息到控制台
                   System.out.println(" 主题： " + 
                   metadata.topic() + "->" + "分区：" + metadata.partition());
                 } else {
        
                   //  出现异常打印
                   exception.printStackTrace();
                  }
                }
             });
         //  延迟一会会看到数据发往不同分区
         Thread.sleep(2);
     }
     // 5. 关闭资源
     kafkaProducer.close();

### 3. 同步发送**API** ###

只需在异步发送的基础上，再调用一下 get()方法即可。

// 同步发送
     kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

### **4.** 生产者分区 ###

#### 4.1 分区好处 ####

（1）**便于合理使用存储资源**，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一

块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。

（2）**提高并行度**，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

![在这里插入图片描述][8b96530c77864c6ba0447e7aa49cbb86.png]

#### 4.2 生产者发送消息的分区策略 ####

**1**）**默认的分区器** **DefaultPartitioner**

在 IDEA 中 ctrl +n，全局查找 DefaultPartitioner。

![在这里插入图片描述][afe33d2c90db4976aba688c470e289cd.png]

#### 4.3 **自定义分区器** ####

**1**）**需求**

例如我们实现一个分区器实现，发送过来的数据中如果包含 xxx，就发往 0 号分区，

不包含 xxx，就发往 1 号分区。

**2**）实现步骤

（1）定义类实现 Partitioner 接口。

（2）重写 partition()方法。

/**
    * 1. 实现接口 Partitioner
    * 2. 实现 3 个方法:partition,close,configure
    * 3. 编写 partition 方法,返回分区号
    */
    public class MyPartitioner implements Partitioner {
        
     /**
     * 返回信息对应的分区
     * @param topic 主题
     * @param key 消息的 key
     * @param keyBytes 消息的 key 序列化后的字节数组
     * @param value 消息的 value
     * @param valueBytes 消息的 value 序列化后的字节数组
     * @param cluster 集群元数据可以查看分区信息
     * @return
     */
     @Override
     public int partition(String topic, Object key, byte[] 
    keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        
        // 获取消息
         String msgValue = value.toString();
         // 创建 partition
         int partition;
         // 判断消息是否包含 atguigu
         if (msgValue.contains("xxx")){
        
         	partition = 0;
         }else {
        
        	 partition = 1;
         }
         // 返回分区号
         return partition;
     }
     // 关闭资源
     @Override
     public void close() {
        
     }
     // 配置方法
     @Override
     public void configure(Map<String, ?> configs) {
        
     } 
    }

（3）使用分区器的方法，在生产者的配置中添加分区器参数。

// 添加自定义分区器
    properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.yy.kafka.producer.MyPartitioner");

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

![在这里插入图片描述][a97ee2a8f4ac4ec99a9fcaed67e329d0.png]

// batch.size：批次大小，默认 16K
     properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
    // linger.ms：等待时间，默认 0
     properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
    // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
     properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
    // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
     properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

### 6. 数据可靠性 ###

1）**ack 应答原理**

![在这里插入图片描述][c31e3184e8fa40c489f8a4901a956e77.png]  
![在这里插入图片描述][d59b59bd967a4b6da5aa17fd7bd43fb4.png]  
![在这里插入图片描述][237b732709874647b6762e103f13286d.png]

// 设置 acks
     properties.put(ProducerConfig.ACKS_CONFIG, "all");
     // 重试次数 retries，默认是 int 最大值，2147483647
     properties.put(ProducerConfig.RETRIES_CONFIG, 3);

### 7. **数据去重** ###

#### **7.1** **数据传递语义** ####

![在这里插入图片描述][848aca831fc9426a9e332a8e32fb3a47.png]

#### 7.2 幂等性 ####

**1**）**幂等性原理**

![在这里插入图片描述][c29bede34d734258a93d494b90aca588.png]

**2**）**如何使用幂等性**

开启参数 **enable.idempotence** 默认为 true，false 关闭。

#### **7.3** **生产者事务** ####

1）Kafka 事务原理

![在这里插入图片描述][007ab35569474534a32037195d88dba3.png]

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;

3）单个 Producer，使用事务保证消息的仅一次发送

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import java.util.Properties;
    public class CustomProducerTransactions {
        
       public static void main(String[] args) throws InterruptedException {
        
         // 1. 创建 kafka 生产者的配置对象
         Properties properties = new Properties();
         // 2. 给 kafka 配置对象添加配置信息
         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                        "hadoop102:9092");
         // key,value 序列化
         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         // 设置事务 id（必须），事务 id 任意起名
         properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
         // 3. 创建 kafka 生产者对象
         KafkaProducer<String, String> kafkaProducer = new 
         KafkaProducer<String, String>(properties);
         // 初始化事务
         kafkaProducer.initTransactions();
         // 开启事务
         kafkaProducer.beginTransaction();
         try {
        
             // 4. 调用 send 方法,发送消息
             for (int i = 0; i < 5; i++) {
        
             // 发送消息
                 kafkaProducer.send(new ProducerRecord<>("first","atguigu " + i));
             }
            // int i = 1 / 0;
             // 提交事务
             kafkaProducer.commitTransaction();
         } catch (Exception e) {
        
             // 终止事务
             kafkaProducer.abortTransaction();
         } finally {
        
             // 5. 关闭资源
             kafkaProducer.close();
         }
     } 
    }

### 8. **数据有序** ###

![在这里插入图片描述][4f4bc428cdfb4f7dabe96098c2428cca.png]

### 9. **数据乱序** ###

1）kafka在1.x版本之前保证数据单分区有序，条件如下：

**max.in.flight.requests.per.connection**=1（不需要考虑是否开启幂等性）。

2）kafka在1.x及以后版本保证数据单分区有序，条件如下：

（1）未开启幂等性

**max.in.flight.requests.per.connection**需要设置为**1**。

（2）开启幂等性

**max.in.flight.requests.per.connection**需要设置小于等于**5**。

原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，

故无论如何，都可以保证最近5个request的数据都是有序的

![在这里插入图片描述][b14d2e0428bc4b44a08193fa6441ee6e.png]

笔记来自[b站尚硅谷][b]

[04aa5562bb664029890ea98a5fbff37b.png]: https://img-blog.csdnimg.cn/04aa5562bb664029890ea98a5fbff37b.png
[7cb451f82a3249a0aa5c732bc41a87a8.png]: https://img-blog.csdnimg.cn/7cb451f82a3249a0aa5c732bc41a87a8.png
[8b96530c77864c6ba0447e7aa49cbb86.png]: https://img-blog.csdnimg.cn/8b96530c77864c6ba0447e7aa49cbb86.png
[afe33d2c90db4976aba688c470e289cd.png]: https://img-blog.csdnimg.cn/afe33d2c90db4976aba688c470e289cd.png
[a97ee2a8f4ac4ec99a9fcaed67e329d0.png]: https://img-blog.csdnimg.cn/a97ee2a8f4ac4ec99a9fcaed67e329d0.png
[c31e3184e8fa40c489f8a4901a956e77.png]: https://img-blog.csdnimg.cn/c31e3184e8fa40c489f8a4901a956e77.png
[d59b59bd967a4b6da5aa17fd7bd43fb4.png]: https://img-blog.csdnimg.cn/d59b59bd967a4b6da5aa17fd7bd43fb4.png
[237b732709874647b6762e103f13286d.png]: https://img-blog.csdnimg.cn/237b732709874647b6762e103f13286d.png
[848aca831fc9426a9e332a8e32fb3a47.png]: https://img-blog.csdnimg.cn/848aca831fc9426a9e332a8e32fb3a47.png
[c29bede34d734258a93d494b90aca588.png]: https://img-blog.csdnimg.cn/c29bede34d734258a93d494b90aca588.png
[007ab35569474534a32037195d88dba3.png]: https://img-blog.csdnimg.cn/007ab35569474534a32037195d88dba3.png
[4f4bc428cdfb4f7dabe96098c2428cca.png]: https://img-blog.csdnimg.cn/4f4bc428cdfb4f7dabe96098c2428cca.png
[b14d2e0428bc4b44a08193fa6441ee6e.png]: https://img-blog.csdnimg.cn/b14d2e0428bc4b44a08193fa6441ee6e.png
[b]: https://space.bilibili.com/302417610?spm_id_from=333.788.b_765f7570696e666f.1