• Kafka - Kafka生产者|发送消息|分区策略|提高吞吐量|数据可靠性|数据去重|数据有序

    文章目录

    1. Kafka 生产者

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

    在这里插入图片描述

    生产者重要参数列表：

    • bootstrap.servers：生产者连接集群所需的broker地址清单，可以设置1个或者多个，中间用逗号隔开。

    • 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 是等价的。

    • retries：当消息发送出现错误的时候，系统会重发消息。retries 表示重试次数。默认是 int 最大值2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送成功了。

    • retry.backoff.ms ：两次重试之间的时间间隔，默认是 100ms

    • enable.idempotence：是否开启幂等性，默认 true，开启幂等性。

    • max.in.flight.requests.per.connection：允许最多没有返回 ack 的次数，默认为 5，开启幂等性
      要保证该值是 1-5 的数字。

    • compression.type：生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。

    2. 生产者发送消息

    1. 生产者异步发送消息

    ① 打开cmd命令行窗口，创建一个主题 topic_test：

    F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat 
                                                --create 
                                                --bootstrap-server localhost:9092 
                                                --replication-factor 1 
                                                -partitions 1 
                                                --topic topic-test

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    ② Maven项目搭建，并导入依赖：

    
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>

    <groupId>com.hhgroupId>
    <artifactId>kafkaartifactId>
    <version>1.0-SNAPSHOTversion>

    <properties>
        <maven.compiler.source>8maven.compiler.source>
        <maven.compiler.target>8maven.compiler.target>
    properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.kafkagroupId>
            <artifactId>kafka-clientsartifactId>
            <version>3.0.0version>
        dependency>
    dependencies>

project>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    ③ kafka 生产者发送消息：

    public class CustomProducer {
    public static void main(String[] args) {

        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者发送消息，默认是异步发送方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            kafkaProducer.send(new ProducerRecord<>("topic-test","message-" + i));
        }
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    ④ cmd命令行窗口开启 kafka 消息者，观察消费者是否接收到消息：

    在这里插入图片描述

    2. 生产者异步发送消息带回调函数

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

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

    public class CustomProducerCallback {
    public static void main(String[] args) throws InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // 带回调函数发送消息
            kafkaProducer.send(new ProducerRecord<>("topic-test", "message-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                    if(exception==null){
                        // 消息发送成功
                        System.out.println("主题"+recordMetadata.topic()+"->"+"分区："+recordMetadata.partition());
                    }else{
                        // 消息发送失败
                        exception.printStackTrace();
                    }
                }
            });
            Thread.sleep(2);
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    在这里插入图片描述

    3. 生产者同步发送消息

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

    public class CustomProducerSync {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者发送消息，默认是异步发送方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // kafka生产者发送消息，同步方式发送
            kafkaProducer.send(new ProducerRecord<>("topic-test","message-" + i)).get();
        }
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    在这里插入图片描述

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

    1. Kafka 分区好处

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

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

    在这里插入图片描述

    2. 分区器 DefaultPartitioner 和 ProducerRecord 源码

    ① 默认的分区器 DefaultPartitioner 源码：

    /**
 * The default partitioning strategy:
 * If a partition is specified in the record, use it
 * If no partition is specified but a key is present choose a partition based on a hash of the key
 * If no partition or key is present choose the sticky partition that changes when the batch is full.
 * 
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {

    private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();

    public void configure(Map<String, ?> configs) {}

    /**
     * Compute the partition for the given record.
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        return partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());
    }

    /**
     * Compute the partition for the given record.
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,
                         int numPartitions) {
        if (keyBytes == null) {
            return stickyPartitionCache.partition(topic, cluster);
        }
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

    public void close() {}
    
    /**
     * If a batch completed for the current sticky partition, change the sticky partition. 
     * Alternately, if no sticky partition has been determined, set one.
     */
    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
        stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43

    ② ProducerRecord 类源码：

    public class ProducerRecord<K, V> {

    private final String topic;
    private final Integer partition;
    private final Headers headers;
    private final K key;
    private final V value;
    private final Long timestamp;

    /**
     * Creates a record with a specified timestamp to be sent to a specified topic and partition
     */
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
        // ...
    }

    /**
     * Creates a record with a specified timestamp to be sent to a specified topic and partition
     */
    public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
        this(topic, partition, timestamp, key, value, null);
    }

    /**
     * Creates a record to be sent to a specified topic and partition
     */
    public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {
        this(topic, partition, null, key, value, headers);
    }
    
    /**
     * Creates a record to be sent to a specified topic and partition
     */
    public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, null, key, value, null);
    }
    
    /**
     * Create a record to be sent to Kafka
     */
    public ProducerRecord(String topic, K key, V value) {
        this(topic, null, null, key, value, null);
    }
    
    /**
     * Create a record with no key
     */
    public ProducerRecord(String topic, V value) {
        this(topic, null, null, null, value, null);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 在指明 partition 的情况下，直接将指明的值作为partition值，如partition=0，则将数据写入分区0；
    • 在没有指明 partition 的情况下，将key的hash值和topic的partition数进行取余得到partition值，例如key的hash值为5，topic的partition数为6，则key对应的value值写入分区1；
    • 既没有指明partition又没有执行key的情况下，afka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同），例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）。
    3. 指定 partition 的情况

    将数据发往指定 partition 的情况。

    修改主题topic-test，增加分区数

    # 修改主题，修改分区数为3
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat --alter --bootstrap-server localhost:9092  --topic topic-test --partitions 3

# 查看主题详情
F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-topics.bat --bootstrap-server localhost:9092 --describe  --topic topic-test

    • 1
    • 2
    • 3
    • 4
    • 5

    在这里插入图片描述

    public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // 构造方法：ProducerRecord(String topic, Integer partition, K key, V value)
            // 将key的value值发送到执行分区1
            kafkaProducer.send(new ProducerRecord<>("topic-test", 1,"","message-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                    if(exception==null){
                        // 消息发送成功
                        System.out.println("主题"+recordMetadata.topic()+"->"+"分区："+recordMetadata.partition());
                    }else{
                        // 消息发送失败
                        exception.printStackTrace();
                    }
                }
            });
            Thread.sleep(2);
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31

    在这里插入图片描述

    4. 没有指定 partition 但指定 key 的情况

    将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值。

    public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // 构造方法：public ProducerRecord(String topic, K key, V value)
            // 依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，分别发往分区 1、2、0
            kafkaProducer.send(new ProducerRecord<>("topic-test", "f","message-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                    if(exception==null){
                        // 消息发送成功
                        System.out.println("主题"+recordMetadata.topic()+"->"+"分区："+recordMetadata.partition());
                    }else{
                        // 消息发送失败
                        exception.printStackTrace();
                    }
                }
            });
            Thread.sleep(2);
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31

    key="ａ"时，在控制台查看结果：

    在这里插入图片描述

    key="ｂ"时，在控制台查看结果：

    在这里插入图片描述

    key="f"时，在控制台查看结果：

    在这里插入图片描述

    5. 既没有指定 partition 也没有指定 key 的情况
    public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");

        // kafka生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // 构造方法：public ProducerRecord(String topic, V value)
            kafkaProducer.send(new ProducerRecord<>("topic-test" ,"message-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                    if(exception==null){
                        // 消息发送成功
                        System.out.println("主题"+recordMetadata.topic()+"->"+"分区："+recordMetadata.partition());
                    }else{
                        // 消息发送失败
                        exception.printStackTrace();
                    }
                }
            });
            Thread.sleep(2);
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    在这里插入图片描述

    6. 自定义分区器

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

    ① 定义类实现 Partitioner 接口：

    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 message= value.toString();
        int partition;
        if(message.contains("chahua")){
            partition = 0;
        }else{
            partition = 1;
        }
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34

    ② 使用分区器的方法，在生产者的配置中添加分区器参数：

    public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws InterruptedException {
        // kafka生产者属性配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");
        // 添加自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.hh.producer.MyPartitioner");

        // kafka生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        for(int i=0;i<5;i++){
            // 构造方法：public ProducerRecord(String topic, V value)
            kafkaProducer.send(new ProducerRecord<>("topic-test" ,"chahua-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                    if(exception==null){
                        // 消息发送成功
                        System.out.println("主题"+recordMetadata.topic()+"->"+"分区："+recordMetadata.partition());
                    }else{
                        // 消息发送失败
                        exception.printStackTrace();
                    }
                }
            });
            Thread.sleep(2);
        }
        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

    控制台查看消息发送的分区：

    在这里插入图片描述

    消费者消费消息：

    F:\install\kafka\kafka_2.12-2.3.0\bin\windows>kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic topic-test --from-beginning
chahua-0
chahua-1
chahua-2
chahua-3
chahua-4

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    4. 生产者如何提高吞吐量

    修改以下参数：

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

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

    • compression.type：压缩snappy

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

    public class CustomProducerParameters {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");
        // batch.size：批次大小，默认为16k
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        // linger.ms：等待时间，默认为0
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
        // RecordAccumulator 缓冲区大小，默认为32M
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
        // 压缩，默认为none，可以配置的值有：gzip、snappy、lz4、zstd
        properties.put(ProducerConfig.CONFIG_PROVIDERS_CONFIG,"snappy");

        // 创建生产者对象
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // send方法发送消息
        for(int i=0;i<5;i++){
            kafkaProducer.send(new ProducerRecord<>("topic-test","hh-"+i));
        }

        // 关闭资源
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    5. 生产者发送数据的可靠性

    ① 生产者发送消息流程：

    在这里插入图片描述

    ② ack 应答原理 ：

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

    在这里插入图片描述

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

    在这里插入图片描述

    ack = -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。
    在这里插入图片描述

    ③ Leader收到数据，所有Follower都开始同步数据，但有一个Follower，因为某种故障，迟迟不能与Leader进行同步，那这个问题怎么解决呢？

    Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合(leader：0，isr:0,1,2)。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如2超时，(leader:0, isr:0,1)。这样就不用等长期联系不上或者已经故障的节点。

    如果分区副本设置为 1个 ，或者ISR里应答的最小副本数量（ min.insync.replicas 默认为1）设置为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，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

    ⑥ 数据重复分析：

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

    在这里插入图片描述

    生产者发送消息Hello，Leader收到消息并同步给了Follower，但是在ack应答时Leader挂掉了，ack失败。于是重新选择一个Leader，由于ack失败，生产者会再发送一个消息Hello给Leader，导致数据重复。

    ⑦ 代码中配置ack应答机制：

    https://kafka.apache.org/documentation/#api

    public class CustomProducerAck {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");
        // 设置acks
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        // 重试次数 retries，默认是int最大值
        properties.put(ProducerConfig.RECEIVE_BUFFER_CONFIG,3);

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        for(int i=0;i<5;i++){
            kafkaProducer.send(new ProducerRecord<>("topic-test","haha"));
        }
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    6. 生产者发送数据去重

    • 至少一次= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2；可以保证数据不丢失，但是不能保证数据不重复。

    • 最多一次= ACK级别设置为0；可以保证数据不重复，但是不能保证数据不丢失。

    • 精确一次：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

    ① 幂等性：

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

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

    重复数据的判断标准：具有相同主键的消息提交时，Broker只会持久化一条。其
    中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。所以幂等性只能保证的是在单分区单会话内不重复。

    在这里插入图片描述

    如何使用幂等性：开启参数 enable.idempotence 默认为 true，false 关闭。

    public class CustomProducerAck {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");
        // 设置acks
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        // 重试次数 retries，默认是int最大值
        properties.put(ProducerConfig.RECEIVE_BUFFER_CONFIG,3);
        // 开启幂等性
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        for(int i=0;i<5;i++){
            kafkaProducer.send(new ProducerRecord<>("topic-test","haha"));
        }
        kafkaProducer.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

    ② 生产者事务：

    尽管开启了幂等性，但是kafka挂掉重启后仍然会导致数据重复，因为其PID在Kafka每次重启都会分配一个新的，因此还需要使用事务。

    说明：开启事务，必须开启幂等性。

    在这里插入图片描述

    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;

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

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

    public class CustomProducerTransactions {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        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");
        // 开启幂等性
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        // 设置事务 id（必须），事务 id 任意起名
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction_id_0");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 初始化事务
        kafkaProducer.initTransactions();
        // 开启事务
        kafkaProducer.beginTransaction();
        try {
            for(int i=0;i<5;i++){
                kafkaProducer.send(new ProducerRecord<>("topic-test","jiang"));
            }
            int i = 1/0;
            // 提交事务
            kafkaProducer.commitTransaction();
        }catch (Exception e){
            // 终止事务
            kafkaProducer.abortTransaction();
        }finally {
            // 关闭资源
            kafkaProducer.close();
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

    7. 生产者发送数据有序

    在这里插入图片描述

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

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

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

    未开启幂等性：max.in.flight.requests.per.connection需要设置为1。

    开启幂等性：max.in.flight.requests.per.connection需要设置小于等于5。

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

    在这里插入图片描述

  • 相关阅读:
    vue源码分析（八）—— update分析(首次渲染)
    uniapp-轮播图点击预览功能
    智能电网中需求响应研究（Matlab代码实现）
    SQL练习题全部
    入门ElasticSearch ：为什么选择ES作为搜索引擎？
    SpringCloudAliBaba篇(三)之Ribbon负载均衡器
    Elasticsearch语法之Term query不区分大小写
    【Java基础面试一】、为什么Java代码可以实现一次编写、到处运行？
    微信小程序开发详细步骤是什么？
    LeetCode--1445. 苹果和桔子
  • 原文地址：https://blog.csdn.net/qq_42764468/article/details/127947191