• Kafka生产者(二)


    1、生产者消息发送流程

    1.1 发送原理

    在消息发送的过程中,涉及到了两个线程——main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator,Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。
    在这里插入图片描述
    分区器是来规定每个数据发往哪个队列(缓存队列),这个队列大小默认是32M。
    Sender线程拉取RecordAccumulator中的消息,需要满足一个条件。batch.size或linger.ms。

    2、异步发送API

    异步通信:允许用户把一个消息放入队列,但并不立即处理它,然后在需要的时候再去处理它们。
    在这里插入图片描述

    2.1 普通异步发送

    1、需求:创建Kafka生产者,采用异步的方式发送到Kafka Broker

    2、代码编写
    (1)创建工程kafka
    (2)导入依赖

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.0.0</version>
        </dependency>
    </dependencies>
    
    2.2 编写不带回调函数的API代码
    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 沉淀、分享、成长,让自己有所收获!
     */
    public class CustomProducer {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 10; i++) {
                KafkaProducer.send(new ProducerRecord<>("first", "aaa:" + i));
            }
    
            // 3、关闭资源
            KafkaProducer.close();
    
        }
    }
    

    测试:
    在192.168.58.130上开启kafka消费者

    root@cl-virtual-machine:/usr/local/kafka# bin/kafka-console-consumer.sh --bootstrap-server 192.168.58.130:9092  --topic first
    

    执行IDEA代码,查看收到发送过来的消息
    在这里插入图片描述

    2.3 带回调函数的异步发送

    回调函数会在 producer 收到 ack 时调用,为异步调用,该方法有两个参数,分别是元数据信息(RecordMetadata)和异常信息(Exception),如果 Exception 为 null,说明消息发送成功,如果 Exception 不为 null,说明消息发送失败。
    在这里插入图片描述

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 沉淀、分享、成长,让自己有所收获!
     */
    public class CustomProducerCallback {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 10; i++) {
                KafkaProducer.send(new ProducerRecord<>("first", "aaa:" + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            System.out.println("topic: " + recordMetadata.topic() + " partition:" + recordMetadata.partition());
                        }
                    }
                });
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    

    3、同步发送API

    同步发送是外部的一批数据发送到双端(缓存)队列的数据,分区的这批数据必须得发送到broker完毕后,再才能发送下一批数据到分区。

    代码:

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    import java.util.concurrent.ExecutionException;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 同步发送
     */
    public class CustomProducerSync {
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 10; i++) {
                KafkaProducer.send(new ProducerRecord<>("first", "aaa:" + i)).get();
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    

    4、生产者-分区器

    4.1 分区好处

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

    (2)提高并行度,生产者可以以分区为单位发送数据;消费者可以以分区为单位进行消费数据,消费者可以组成一个消费者组,去并行的消费数据,提高消费能力。
    在这里插入图片描述
    将100T的数据分为三份,每份是33T分别存放到三台服务器上,这样可以减少服务器的存储压力。

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

    1、默认的分区器DefaultPartitioner
    在这里插入图片描述

    2、案例一
    将数据发送到指定的partition的情况下,例如,将所有数据发往分区 1 中。

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 指定分区发送生产者数据
     */
    public class CustomProducerCallbackPartition {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 5; i++) {
                //  指定数据发送到 1 号分区,key 为空
                KafkaProducer.send(new ProducerRecord<>("first",1,"", "aaa:" + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            System.out.println("topic: " + recordMetadata.topic() + " partition:" + recordMetadata.partition());
                        }
                    }
                });
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    

    以下蓝色部分是新增的数据。
    在这里插入图片描述

    3、案例二
    没有指明 partition 值但有 key 的情况下,将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值。

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 指定分区发送生产者数据
     */
    public class CustomProducerCallbackPartition {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 5; i++) {
                // 依次指定 key 值为 a,b,f ,数据 key 的 hash 值与 3 个分区求余,
                //分别发往 1、2、0
                KafkaProducer.send(new ProducerRecord<>("first","f", "aaa:" + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            System.out.println("topic: " + recordMetadata.topic() + " partition:" + recordMetadata.partition());
                        }
                    }
                });
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    
    

    在这里插入图片描述

    4.3 自定义分区器

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

    2、实现步骤
    (1)定义类实现Partitioner 接口。
    (2)重写 partition()方法。

    package com.atguigu.kafka.config;
    
    import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    
    import java.util.Map;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 自定义分区器
     */
    public class MyPartition implements Partitioner {
        @Override
        public int partition(String topic, Object key, byte[] bytes, Object value, byte[] bytes1, Cluster cluster) {
            // 获取数据 atguigu hello
            String msgValues = value.toString();
            int partition;
            if (msgValues.contains("atguigu")) {
                partition = 0;
            } else {
                partition = 1;
            }
            return partition;
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void configure(Map<String, ?> map) {
    
        }
    }
    
    

    (3)使用分区器的方法,在生产者的配置中添加分区器参数。

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.*;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 指定分区发送生产者数据
     */
    public class CustomProducerCallbackPartition {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 配置自定义分区
            properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.config.MyPartition");
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 5; i++) {
                // 包含atguigu 发往0号分区,否则发往1号分区
                KafkaProducer.send(new ProducerRecord<>("first","hello" + i), new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            System.out.println("topic: " + recordMetadata.topic() + " partition:" + recordMetadata.partition());
                        }
                    }
                });
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    
    
    4.4 生产者如何提高吞吐量

    在这里插入图片描述

    package com.atguigu.kafka.producer;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: 提高生产者吞吐量的相关参数配置
     */
    public class CustomProducerParameters {
    
        public static void main(String[] args) {
    
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 1.3 缓冲区大小
            properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
            // 1.4 批次大小
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
            // 1.5 linger.ms   1ms
            properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
            // 1.6 压缩
            properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 5; i++) {
                KafkaProducer.send(new ProducerRecord<>("first", "aaa:" + i));
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    
    

    测试,消费者收到发送过来的数据
    在这里插入图片描述

    4.5 生产者-数据的可靠性

    ACK应答级别
    (1)acks= 0
    在这里插入图片描述
    (2)acks= 1
    在这里插入图片描述
    (3)acks= -1

    在这里插入图片描述
    Follower会主动向Leader拉取数据。

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

    在这里插入图片描述
    在这里插入图片描述

    可靠性总结:
    acks=0,生产者发送过来数据就不管了,可靠性差,效率高;
    acks=1,生产者发送过来数据Leader应答,可靠性中等,效率中等;
    acks=-1,生产者发送过来数据Leader和ISR队列里面所有Follwer应答,可靠性高,效率低;
    在生产环境中,acks=0很少使用;acks=1,一般用于传输普通日志,允许丢个别数据;acks=-1,一般用于传输和钱相关的数据,对可靠性要求比较高的场景。

    数据重复性分析:
    acks: -1(all):生产者发送过来的数据,Leader和ISR队列里面的所有节点收齐数据后应答。

    在这里插入图片描述
    当Leader向生产者发送ack应答时,刚好Leader挂了,由于生产者没有收到ACK应答,会认为没有消费到,所以Hello这条数据不会被清除。当其他的Follower升级为Leader时,由于之前这个Follower已经将Hello这条数据同步过来了,现在升级后,生产者还会将Hello这条数据再发送一遍,就会导致接收到两份Hello数据,导致数据重复。

    代码:

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright:设置ack应答机制、生产者重试次数
     */
    public class CustomProducerAcks {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的key和value的序列化类型
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // ACKS
            properties.put(ProducerConfig.ACKS_CONFIG, "1");
            // 重试次数
            properties.put(ProducerConfig.RETRIES_CONFIG, 3);
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
            // 2.1 发送数据
            for (int i = 0; i < 10; i++) {
                KafkaProducer.send(new ProducerRecord<>("first", "bbb:" + i));
            }
    
            // 3、关闭资源
            KafkaProducer.close();
        }
    }
    
    4.6 生产经验 - 数据去重

    1、数据传递语义

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

    • 最多一次(At Most Once)= ACK级别设置为0

    • 总结:
      At Least Once可以保证数据不丢失,但是不能保证数据不重复;
      At Most Once可以保证数据不重复,但是不能保证数据不丢失。

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

    4.7 幂等性

    1、幂等性原理
    幂等性就是指Producer不论向Broker发送多少次重复数据,Broker端都只会持久化一条,保证了不重复。
    精确一次(Exactly Once) = 幂等性 + 至少一次( ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2) 。

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

    在这里插入图片描述
    2、如何使用幂等性
    开启参数 enable.idempotence 默认为 true,false 关闭。

    4.8 生产者事务

    1、Kafka事务原理
    在这里插入图片描述
    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,使用事务保证消息的仅一次发送

    package com.atguigu.kafka.producer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    /**
     * @description:
     * @author:cl
     * @date: 2024/8/4
     * @Copyright: Kafka事务
     */
    public class CustomProducerTransactions {
    
        public static void main(String[] args) {
            // 1、配置
            Properties properties = new Properties();
            // 1.1 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.58.130:9092");
            // 1.2 指定对应的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");
    
    
            // 2.创建kafka生产者对象
            KafkaProducer<String, String> KafkaProducer = new KafkaProducer<>(properties);
    
            // 初始化事务
            KafkaProducer.initTransactions();
            // 开启事务
            KafkaProducer.beginTransaction();
    
            try {
                // 2.1 发送数据
                for (int i = 0; i < 10; i++) {
                    KafkaProducer.send(new ProducerRecord<>("first", "aaa:" + i));
                }
                int i = 10/0;
                // 提交事务
                KafkaProducer.commitTransaction();
            } catch (Exception e) {
                // 终止事务
                KafkaProducer.abortTransaction();
            } finally {
                // 3、关闭资源
                KafkaProducer.close();
            }
        }
    }
    
  • 相关阅读:
    甲子光年专访天润融通CEO吴强:客户经营如何穿越低速周期?
    Verilog编写VGA控制器
    Nginx企业级负载均衡:技术详解系列(17)—— 长连接优化策略与下载服务器高效搭建
    案例分享-https证书链不完整导致请求失败
    Mysql查询去空格方法汇总
    如何用Know Streaming来查询Kafka的消息
    Emacs之定制化mode line(第一百零二)
    Apollo 应用与源码分析:Monitor监控-软件监控-定位、camera、功能安全、数据记录监控
    Pandas将三个聚合结果的列,如何合并到一张表里?
    Appium 全新 2.0 全新跨平台生态,版本特性抢鲜体验!
  • 原文地址:https://blog.csdn.net/java123456111/article/details/140907629