• kafka笔记（二）：生产者-同步发送/异步发送/生产者分区/数据可靠性/数据去重/消息发送流程

    目录

    kafka生产者

    生产者消息发送流程

    发送原理

    生产者重要参数列表

    异步发送API

    普通异步发送

    带回调函数的异步发送

    同步发送API

    生产者分区

    分区好处

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

    自定义分区器

    生产经验—提高生产者的吞吐量

    生产经验—数据可靠性

    生产经验—数据去重

    数据传递语义

    幂等性

    生产者事务

    生产经验—数据有序

    生产经验—数据乱序

    kafka生产者

    生产者消息发送流程

    发送原理

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

    batch.size：只有数据积累到batch.size之后，sender才会发送数据。默认16k。

    linger.ms：如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。

    应答acks：

    0：生产者发送过来的数据，不需要等数据落盘应答。 

    1：生产者发送过来的数据，Leader收到数据后应答。

    -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。-1和all等价。

    生产者重要参数列表

    参数名称描述
    bootstrap.servers

    生产者连接集群所需的broker地址清单。例如：hadoop01:9092，hadoop02:9092，hadoop03:9092，可以设置1个或者多个，中间用逗号隔开。注意，这里并非需要所有的broker地址，因为生产者从给定的broker里查找到其他broker信息。

    key.serializer和value.serializer

    指定发送消息的key和value的序列化类型。（注：要写全列名）

    buffer.memory

    RecordAccumulator缓冲区总大小，默认32M。

    batch.size

    缓冲区一批数据的最大值，默认16k。适当增加该值可以提高吞吐量，但是如果该值设置太大会导致数据传输延迟增加。

    linger.ms

    如果数据迟迟未达到batch.size，sender等待linger.time设置的时间到了就会发送数据。单位ms，默认值是0ms：表示没有延迟。

    生产环境建议该值大小为5-100ms之间。

    acks

    0：生产者发送过来的数据，不需要等数据落盘应答。

    1：生产者发送过来的数据，Leader收到数据后应答。

    -1（all）：生产者发送过来的数据，Leader和isr队列里面的所有节点收齐数据后应答。默认值是-1，-1和all是等价的。

    max.in.flight.requests.per.connection

    允许最多没有返回ack的次数，默认为5，开启幂等性要保证该值是1-5的数字。

    retry.backoff.ms

    两次重试之间的时间间隔，默认是100ms。

    enable.idempotence

    是否开启幂等性，默认true，开启幂等性。

    compression.type

    生产者发送的所有数据的压缩方式。默认是none，也就是不压缩。

    支持压缩类型：none、gzip、snappy、lz4和zstd。

    retries

    当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是int最大值：2147483647。

    如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。

    异步发送API

    普通异步发送

    （1）需求：创建kafka生产者，采用异步发送的方式发送到kafka broker。

    （2）代码编写：

    1）创建maven工程

    2）导入依赖

    1.     <dependencies>
    2.         <dependency>
    3.             <groupId>org.apache.kafka</groupId>
    4.             <artifactId>kafka-clients</artifactId>
    5.             <version>2.4.1</version>
    6.         </dependency>
    7.     </dependencies>
    8.  
    9.     <build>
    10.         <plugins>
    11.             <plugin>
    12.                 <artifactId>maven-compiler-plugin</artifactId>
    13.                 <version>3.6.0</version>
    14.                 <configuration>
    15.                     <source>1.8</source>
    16.                     <target>1.8</target>
    17.                 </configuration>
    18.             </plugin>
    19.             <plugin>
    20.                 <artifactId>maven-assembly-plugin</artifactId>
    21.                 <configuration>
    22.                     <descriptorRefs>
    23.                         <descriptorRef>jar-with-dependencies</descriptorRef>
    24.                     </descriptorRefs>
    25.                 </configuration>
    26.                 <executions>
    27.                     <execution>
    28.                         <id>make-assembly</id>
    29.                         <phase>package</phase>
    30.                         <goals>
    31.                             <goal>single</goal>
    32.                         </goals>
    33.                     </execution>
    34.                 </executions>
    35.             </plugin>
    36.         </plugins>
    37.     </build>

    3）创建包

    4）编写API代码

    1. import org.apache.kafka.clients.producer.KafkaProducer;
    2. import org.apache.kafka.clients.producer.ProducerConfig;
    3. import org.apache.kafka.clients.producer.ProducerRecord;
    4.  
    5. import java.util.Properties;
    6.  
    7. public class CustomProducer {
    8.     public static void main(String[] args) {
    9.         /**
    10.          编写不带回调函数的API代码
    11.          */
    12.         //1.创建kafka生产者的配置对象
    13.         Properties properties = new Properties();
    14.         //2.给kafka配置对象添加配置信息：bootstrap.server
    15.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    16.         //key，value序列化：key.serializer，value.serializer
    17.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    18.                 "org.apache.kafka.common.serialization.StringSerializer");
    19.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    20.                 "org.apache.kafka.common.serialization.StringSerializer");
    21.         //3.创建kafka生产者对象
    22.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    23.         //4.调用send方法，发送信息
    24.         for (int i = 0;i < 5;i++) {
    25.             kafkaProducer.send(new ProducerRecord<>("first","hello kafka" + i));
    26.         }
    27.         //5.关闭资源
    28.         kafkaProducer.close();
    29.     }
    30. }

    （3）测试

    1）开启kafka

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

    2）执行idea的代码，查看kafka是否接收到消息

    带回调函数的异步发送

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

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

    （1）代码编写

    1. import org.apache.kafka.clients.producer.*;
    2. import org.apache.kafka.common.serialization.StringSerializer;
    3.  
    4. import java.util.Properties;
    5.  
    6. public class CustomProducerCallBack {
    7.     public static void main(String[] args) throws InterruptedException {
    8.         /**
    9.          * 编写回调函数异步发送API代码
    10.          */
    11.         //1.创建kafka生产者环境
    12.         Properties properties = new Properties();
    13.         //2.给kafka配置对象添加配置信息：bootstrap.server
    14.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    15.         //key.value序列化：key.serializer,value.serializer
    16.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    17.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    18.         //3.创建kafka生产者对象
    19.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    20.         //4.调用send方法，发送信息
    21.         for (int a = 0;a < 5; a++) {
    22.             kafkaProducer.send(new ProducerRecord<>("first", "hello mykafka" + a), new Callback() {
    23.                 //在kafkaProducer接收到ack时调用，异步调用
    24.                 @Override
    25.                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {
    26.                     if (e == null) {
    27.                         System.out.println("主题："+ recordMetadata.topic() + "->"+"分区:"+recordMetadata.partition());
    28.                     } else {
    29.                         //出现异常打印
    30.                         e.printStackTrace();
    31.                     }
    32.                 }
    33.             });
    34.             //延迟一会会看到数据发往不同的分区
    35.             Thread.sleep(2);
    36.         }
    37.         //5.关闭环境
    38.         kafkaProducer.close();
    39.     }
    40. }

    （2）测试

    1）开启消费者

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

    2）执行idea的代码，查看控制台中是否有接收到回调消息

    3）查看kafka是否接收到消息

    同步发送API

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

    （1）代码编写

    1. import org.apache.kafka.clients.producer.KafkaProducer;
    2. import org.apache.kafka.clients.producer.ProducerConfig;
    3. import org.apache.kafka.clients.producer.ProducerRecord;
    4. import org.apache.kafka.common.serialization.StringSerializer;
    5.  
    6. import java.util.Properties;
    7. import java.util.concurrent.ExecutionException;
    8.  
    9. public class CustomProducerS {
    10.     public static void main(String[] args) throws ExecutionException,InterruptedException {
    11.         /**
    12.          * 同步发送API
    13.          */
    14.         //1.创建kafka生产者的配置环境
    15.         Properties properties = new Properties();
    16.         //2.给kafka配置对象添加配置信息
    17.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    18.         //key，value序列化：key.serializer，value.serializer
    19.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    20.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    21.         //3.创建kafka生产者对象
    22.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    23.         //4.调用send方法，发送消息
    24.         for (int i = 0; i < 8; i++) {
    25.             //异步发送（默认）
    26.             //kafkaProducer.send(new ProducerRecord<>("first","hello!"+i));
    27.             //同步发送
    28.             kafkaProducer.send(new ProducerRecord<>("first","kafka"+i)).get();
    29.         }
    30.         //5.关闭资源
    31.         kafkaProducer.close();
    32.     }
    33. }

    （2）测试

    1）开启消费者

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

    2）执行idea的代码，查看是否有接收到消息

    生产者分区

    分区好处

    （1）便于合理使用存储资源：每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。

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

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

    （1）默认分区器：DefaultPartitioner

    1）指明partition的情况下，直接将指明的值作为partition值；

    例如partition=0，所有数据写入分区0。

    2）没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；

    例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那

    么key1对应的value1写入1号分区，key2对应的value2写入0号分区。

    3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器）随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。

    例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）。

    （2）案例一

    将数据发送指定partition，如：将数据指定发送分区0中。

    1）代码

    1. import org.apache.kafka.clients.producer.*;
    2. import org.apache.kafka.common.serialization.StringSerializer;
    3.  
    4. import java.util.Properties;
    5. import java.util.concurrent.ExecutionException;
    6.  
    7. public class CustomProducerPartitions {
    8.     public static void main(String[] args) throws ExecutionException,InterruptedException{
    9.         /**
    10.          * 将数据发送到指定分区
    11.          * 将所有数据发送到分区0中
    12.          */
    13.         //1.创建kafka生产者的配置对象
    14.         Properties properties = new Properties();
    15.         //2.给kafka配置对象添加配置信息
    16.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    17.         //key，value序列化：key.serializer，value.serializer
    18.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    19.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    20.         //3.创建kafka生产者对象
    21.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    22.         //4.调用send方式，方式消息
    23.         for (int i = 0; i < 5; i++) {
    24.             //指定数据发送到1分区
    25.             kafkaProducer.send(new ProducerRecord<>("first", 0, "", "hi spark" + i), new Callback() {
    26.                 @Override
    27.                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {
    28.                     if (e == null) {
    29.                         System.out.println("主题：" + recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());
    30.                     } else {
    31.                         e.printStackTrace();
    32.                     }
    33.                 }
    34.             });
    35.         }
    36.         //5.关闭环境
    37.         kafkaProducer.close();
    38.     }
    39. }

    2）测试

    开启kafka消费者

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

    执行idea中代码，观察控制台和kafka的消息

    控制台

    kafka

    （3）案例二

    没有指定partition值但是有key的情况，将key的hash值与topic的partition数进行取余得到partition值

    1）代码

    1. import org.apache.kafka.clients.producer.*;
    2. import org.apache.kafka.common.serialization.StringSerializer;
    3.  
    4. import java.util.Properties;
    5.  
    6. public class CustomProducerPartitions2 {
    7.     public static void main(String[] args) {
    8.         /**
    9.          * 没有指定partition值但是有key的情况
    10.          * 将key的hash值与topic的partition数进行取余得到partition值
    11.          */
    12.         //1.创建kafka生产者的配置对象
    13.         Properties properties = new Properties();
    14.         //2.给kafka配置对象添加配置信息
    15.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    16.         //key，value序列化：key.serializer，value.serializer
    17.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    18.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    19.         //3.创建kafka生产者对象
    20.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    21.         //4.调用send方式，方式消息
    22.         for (int i = 0; i < 5; i++) {
    23.             //
    24.             kafkaProducer.send(new ProducerRecord<>("first", "a","world" + i), new Callback() {
    25.                 @Override
    26.                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {
    27.                     if (e == null) {
    28.                         System.out.println("主题：" + recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());
    29.                     } else {
    30.                         e.printStackTrace();
    31.                     }
    32.                 }
    33.             });
    34.         }
    35.         //5.关闭环境
    36.         kafkaProducer.close();
    37.     }
    38. }

    2）测试

    开启kafka消费者

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

    key=a        查看控制台结果

    key=b        查看控制台结果

    key=c        查看控制台结果

     key=f        查看控制台结果

    自定义分区器

    根据需求可以自己重新实现分区器

    （1）需求：实现一个分区器，发送的数据出现jeffry就发送到分区1，否则发送到分区2.

    （2）实现：

    1）代码

    partition：

    1. import org.apache.kafka.clients.producer.*;
    2. import org.apache.kafka.common.Cluster;
    3.  
    4. import java.util.Map;
    5. /**
    6.  * 1.实现接口partitioner
    7.  * 2.实现三个方法：partition，close，configure
    8.  * 3.编写partition方法，返回分区号
    9.  */
    10. public class MyPartition implements Partitioner {
    11.     /**
    12.      *
    13.      * @param topic     主题
    14.      * @param key       消息的key
    15.      * @param keyBytes  消息的key序列化后的字节数组
    16.      * @param value     消息的value
    17.      * @param valueBytes 消息的value序列化后的字节数组
    18.      * @param cluster    集群元数据可以查看分区信息
    19.      * @return
    20.      */
    21.  
    22.     @Override
    23.     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    24.         //获取消息
    25.         String msgValue = value.toString();
    26.         //创建partition
    27.         int partition;
    28.         //判断消息是否包含jeffry
    29.         if (msgValue.contains("jeffry")) {
    30.             partition = 1;
    31.         } else {
    32.             partition = 2;
    33.         }
    34.  
    35.         //返回分区号
    36.         return partition;
    37.     }
    38.  
    39.     //关闭资源
    40.     @Override
    41.     public void close() {
    42.  
    43.     }
    44.  
    45.     //配置方法
    46.     @Override
    47.     public void configure(Map configs) {
    48.  
    49.     }
    50. }

    测试自定义partition：

    1. import org.apache.kafka.clients.producer.*;
    2. import org.apache.kafka.common.serialization.StringSerializer;
    3.  
    4. import java.util.Properties;
    5.  
    6. public class CustomProducerMyPartitions {
    7.     public static void main(String[] args) {
    8.         /**
    9.          * 测试自定义partition
    10.          */
    11.         Properties properties = new Properties();
    12.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    13.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    14.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    15.  
    16.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.zj.kafka.MyPartition");
    17.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    18.         for (int i = 0;i < 5;i++) {
    19.             kafkaProducer.send(new ProducerRecord<>("first", "jeffry" + i), new Callback() {
    20.                 @Override
    21.                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {
    22.                     if (e == null) {
    23.                         System.out.println("主题：" + recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());
    24.                     } else {
    25.                         e.printStackTrace();
    26.                     }
    27.                 }
    28.             });
    29.         }
    30.         kafkaProducer.close();
    31.     }
    32. }

    2）测试

    开启kafka消费者

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

    在idea启动执行程序

    查看控制台的信息

     查看kafka的信息

     

    生产经验—提高生产者的吞吐量

    batch.size：批次大小，默认16k；

    linger.ms：等待时间，修改为5-100ms；

    compression.type：压缩形式为snappy；

    RecordAccumulator：缓冲区大小，修改为64m；

    （1）代码

    1. import org.apache.kafka.clients.producer.KafkaProducer;
    2. import org.apache.kafka.clients.producer.ProducerConfig;
    3. import org.apache.kafka.clients.producer.ProducerRecord;
    4. import org.apache.kafka.common.serialization.StringSerializer;
    5.  
    6. import java.util.Properties;
    7.  
    8. public class Parameters {
    9.     public static void main(String[] args) {
    10.         //创建kafka生产者配置对象
    11.         Properties properties = new Properties();
    12.         //给kafka配置对象添加配置信息
    13.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    14.         //key，value序列化
    15.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    16.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    17.  
    18.         //batch.size：批次大小，默认16K
    19.         properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
    20.         //linger.ms：等待时间，默认0
    21.         properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
    22.         //RecordAccumulator：缓冲区大小，默认32M buffer。memory
    23.         properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
    24.         //compression.type：压缩，默认none，可以配置gzip、snappy、lzo、zstd
    25.         properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
    26.  
    27.         //创建kafka生产者对象
    28.         KafkaProducer kafkaProducer = new KafkaProducer<>(properties);
    29.         //调用send方法，发送信息
    30.         for (int i = 0;i < 5;i++) {
    31.             kafkaProducer.send(new ProducerRecord<>("first","jeffryOne" + i));
    32.         }
    33.         //关闭环境
    34.         kafkaProducer.close();
    35.     }
    36. }

    （2）测试

    1）开启kafka消费者

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

    2）在idea中启动执行程序

    观察kafka是否接收到消息

    生产经验—数据可靠性

    （1）ack应答原理

     

    数据可靠性分析：

    如果分区副本设置为1个，或者ISR里应答的最小副本数量设置为1（ min.insync.replicas默认为1，和ack=1的效果是一样的，仍然有丢数的风险（leader：0，isr:0）。

    数据完全可靠条件 = ACK级别为-1+分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

    可靠性总结：

    acks=0，生产者发送过来数据就不管了，可靠性差，效率高；

    acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；

    acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；

    在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

    数据重复分析：

    acks： -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。

    （2）代码编写

    1. import org.apache.kafka.clients.producer.KafkaProducer;
    2. import org.apache.kafka.clients.producer.ProducerConfig;
    3. import org.apache.kafka.clients.producer.ProducerRecord;
    4. import org.apache.kafka.common.serialization.StringSerializer;
    5.  
    6. import java.util.Properties;
    7.  
    8. public class CustomProducerAck {
    9.     public static void main(String[] args) {
    10.         //创建kafka生产者的配置对象
    11.         Properties properties = new Properties();
    12.         //给kafka配置对象添加配置信息
    13.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    14.         //key，value序列化
    15.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    16.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    17.  
    18.         //设置acks
    19.         properties.put(ProducerConfig.ACKS_CONFIG,"all");
    20.         //重试次数retries。默认是int最大的值：2147483647
    21.         properties.put(ProducerConfig.RETRIES_CONFIG,5);
    22.  
    23.         //创建kafka生产者对象
    24.         KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
    25.         //调用send方法，发送信息
    26.         for (int i = 0;i < 5;i++) {
    27.             kafkaProducer.send(new ProducerRecord<>("first","jeffry" + i));
    28.         }
    29.         //关闭环境
    30.         kafkaProducer.close();
    31.     }
    32. }

    （3）测试

    1）开启kafka消费者

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

    2）在idea中启动执行程序，查看kafka情况

    生产经验—数据去重

    数据传递语义

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

    At Least Once可以保证数据不丢失，但是不能保证数据不重复。

    （2）最多一次（At Most Once）= ACK级别设置为0；

    At Most Once可以保证数据不重复，但是不能保证数据不丢失。

    （3）精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

    幂等性

    （1）幂等性原理

    幂等性：指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。

    精确一次 = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数>=2）。

    判断重复数据的标准：具有相同主键的消息提交时，Broker只会持久化一条。

    PID是Kafka每次重启都会分配一个新的；

    Partition表示分区号；

    Sequence Number是单调自增的。

    幂等性只能保证在单分区单会话内不重复。

    （2）如何使用幂等性

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

    生产者事务

    （1）kafka事务原理

    注：开启事务必须开启幂等性。 

     （2）kafka事务的5个API

    1. //1初始化事务
    2. void initTransactions();
    3. //2开启事务
    4. void beginTransaction() throws ProducerFencedException;
    5. //3在事务内提交已经消费的偏移量
    6. void sendOffsetsToTransaction(Map offsets,
    7.        String consumerGroupId) throws ProducerFencedException;
    8. //4提交事务
    9. void commitTransaction() throws ProducerFencedException;
    10. //5放弃事务
    11. void abortTransaction() throws ProducerFencedException;

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

    1. import org.apache.kafka.clients.producer.KafkaProducer;
    2. import org.apache.kafka.clients.producer.ProducerConfig;
    3. import org.apache.kafka.clients.producer.ProducerRecord;
    4. import org.apache.kafka.common.serialization.StringSerializer;
    5.  
    6. import java.util.Properties;
    7.  
    8. public class Transaction {
    9.     public static void main(String[] args) throws InterruptedException {
    10.         //1.创建kafka生产者的配置对象
    11.         Properties properties = new Properties();
    12.         //2.给kafka配置对象添加配置信息
    13.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop01:9092,hadoop02:9092");
    14.         //key，value序列化
    15.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    16.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    17.  
    18.         //设置事务id（必须），事务id可以任意起名
    19.         properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transactionis-0");
    20.  
    21.         //3.创建kafka生产者对象
    22.         KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
    23.         //初始化事务
    24.         kafkaProducer.initTransactions();
    25.         //开启事务
    26.         kafkaProducer.beginTransaction();
    27.         try {
    28.             //4.调用send方法，发送信息
    29.             for (int i = 0;i < 5;i++) {
    30.                 //发送消息
    31.                 kafkaProducer.send(new ProducerRecord<>("first","tom" + i));
    32.             }
    33.             int i = 1/0;
    34.             //提交事务
    35.             kafkaProducer.commitTransaction();
    36.         } catch (Exception e) {
    37.             //中止事务
    38.             kafkaProducer.abortTransaction();
    39.         } finally {
    40.             //5.关闭环境
    41.             kafkaProducer.close();
    42.         }
    43.     }
    44. }

    生产经验—数据有序

    单分区内数据是有序的；

    多发区内数据是无序的；

    生产经验—数据乱序

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

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

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

    1）开启幂等性

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

    2）未开启幂等性

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

    原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。    

    如果开启了幂等性且缓存的请求个数小于5个。会在服务端重新排序。

    本文为学习笔记！！！

  • 相关阅读:
    电子元器件行业B2B交易系统：规范企业交易流程，提升销售管理效率
    Cadence Allegro 在Gerber光绘中生成板卡层叠结构文件
    【Qt】Qt中的中心部件意义
    Elasticsearch 注册为Windows服务
    ValueError: Unknown engine: openpyxl，pandas指定读取新版本execl
    【NLP】python进行word文档编辑——构建不同层级标题
    excel表的筛选后自动求和
    开服当GM的基本准则
    linux中busybox与文件系统的关系
    2023年第十六届山东省职业院校技能大赛中职组“网络安全”赛项规程
  • 原文地址：https://blog.csdn.net/qq_55906442/article/details/126857953