0
0

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代码
packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 沉淀、分享、成长，让自己有所收获！
 */publicclassCustomProducer{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <10; i++){KafkaProducer.send(newProducerRecord<>("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，说明消息发送失败。
在这里插入图片描述

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.*;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 沉淀、分享、成长，让自己有所收获！
 */publicclassCustomProducerCallback{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <10; i++){KafkaProducer.send(newProducerRecord<>("first","aaa:"+ i),newCallback(){@OverridepublicvoidonCompletion(RecordMetadata recordMetadata,Exception e){if(e ==null){System.out.println("topic： "+ recordMetadata.topic()+" partition："+ recordMetadata.partition());}}});}// 3、关闭资源KafkaProducer.close();}}

3、同步发送API

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

代码：

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;importjava.util.concurrent.ExecutionException;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 同步发送
 */publicclassCustomProducerSync{publicstaticvoidmain(String[] args)throwsExecutionException,InterruptedException{// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <10; i++){KafkaProducer.send(newProducerRecord<>("first","aaa:"+ i)).get();}// 3、关闭资源KafkaProducer.close();}}

4、生产者-分区器

4.1 分区好处

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

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

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

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

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

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.*;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 指定分区发送生产者数据
 */publicclassCustomProducerCallbackPartition{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <5; i++){//  指定数据发送到 1 号分区，key 为空KafkaProducer.send(newProducerRecord<>("first",1,"","aaa:"+ i),newCallback(){@OverridepublicvoidonCompletion(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 值。

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.*;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 指定分区发送生产者数据
 */publicclassCustomProducerCallbackPartition{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <5; i++){// 依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，//分别发往 1、2、0KafkaProducer.send(newProducerRecord<>("first","f","aaa:"+ i),newCallback(){@OverridepublicvoidonCompletion(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()方法。

packagecom.atguigu.kafka.config;importorg.apache.kafka.clients.producer.Partitioner;importorg.apache.kafka.common.Cluster;importjava.util.Map;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 自定义分区器
 */publicclassMyPartitionimplementsPartitioner{@Overridepublicintpartition(String topic,Object key,byte[] bytes,Object value,byte[] bytes1,Cluster cluster){// 获取数据 atguigu helloString msgValues = value.toString();int partition;if(msgValues.contains("atguigu")){
            partition =0;}else{
            partition =1;}return partition;}@Overridepublicvoidclose(){}@Overridepublicvoidconfigure(Map<String,?> map){}}

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

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.*;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 指定分区发送生产者数据
 */publicclassCustomProducerCallbackPartition{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <5; i++){// 包含atguigu 发往0号分区，否则发往1号分区KafkaProducer.send(newProducerRecord<>("first","hello"+ i),newCallback(){@OverridepublicvoidonCompletion(RecordMetadata recordMetadata,Exception e){if(e ==null){System.out.println("topic： "+ recordMetadata.topic()+" partition："+ recordMetadata.partition());}}});}// 3、关闭资源KafkaProducer.close();}}
4.4 生产者如何提高吞吐量

在这里插入图片描述

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： 提高生产者吞吐量的相关参数配置
 */publicclassCustomProducerParameters{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <5; i++){KafkaProducer.send(newProducerRecord<>("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数据，导致数据重复。

代码：

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright：设置ack应答机制、生产者重试次数
 */publicclassCustomProducerAcks{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 2.1 发送数据for(int i =0; i <10; i++){KafkaProducer.send(newProducerRecord<>("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） 。

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

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

4.8 生产者事务

1、Kafka事务原理
在这里插入图片描述
2、Kafka的事务一共有如下5个API

// 1 初始化事务voidinitTransactions();// 2 开启事务voidbeginTransaction()throwsProducerFencedException;// 3 在事务内提交已经消费的偏移量（主要用于消费者）voidsendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata> offsets,String consumerGroupId)throwsProducerFencedException;// 4 提交事务voidcommitTransaction()throwsProducerFencedException;// 5 放弃事务（类似于回滚事务的操作）voidabortTransaction()throwsProducerFencedException;

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

packagecom.atguigu.kafka.producer;importorg.apache.kafka.clients.producer.KafkaProducer;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.apache.kafka.common.serialization.StringSerializer;importjava.util.Properties;/**
 * @description:
 * @author：cl
 * @date: 2024/8/4
 * @Copyright： Kafka事务
 */publicclassCustomProducerTransactions{publicstaticvoidmain(String[] args){// 1、配置Properties properties =newProperties();// 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=newKafkaProducer<>(properties);// 初始化事务KafkaProducer.initTransactions();// 开启事务KafkaProducer.beginTransaction();try{// 2.1 发送数据for(int i =0; i <10; i++){KafkaProducer.send(newProducerRecord<>("first","aaa:"+ i));}int i =10/0;// 提交事务KafkaProducer.commitTransaction();}catch(Exception e){// 终止事务KafkaProducer.abortTransaction();}finally{// 3、关闭资源KafkaProducer.close();}}}
标签： kafka linq 分布式
本文转载自: https://blog.csdn.net/java123456111/article/details/140907629
版权归原作者 程序员老石 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Kafka生产者（二）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

Visual Studio Code(VS Code)配置C/C++环境

RabbitMQ高效的消息队列中间件原理及实践

RabbitMQ 发布确认高级部分

oracle和hive之间关于sql的语法差异及转换

数据库基础（MySQL）

MySQL：库和表的操作

如何安装Rocky Linux8.5？详细过程及安装包如下

文章导航