当前位置：首页 > news >正文

KAFKA第二课之生产者(面试重点)

news 2025/8/3 2:19:43

生产者学习

1.1 生产者消息发送流程

在消息发送的过程中，涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。
生产者如何发送的？
现在Main线程中将数据进行处理，处理成IO型数据，然后调用sender进行发送
Main:
1.读取生产者配置
2.产生数据
3.过滤数据（校验什么的）
4.序列化
5.放入缓冲区 RecordAccumulator
6.发送Sender

细节：考虑的问题 1.生产者配置的读取和修改 2.数据的过滤与分区， 3.缓冲区是如何设置的，大小
4.发送（发送失败怎么样，请求区的大小）
这里注意一下，可以在缓冲区对数据进行压缩，这样就提高缓冲区的容量和发送的数据量，提高吞吐量

1.2 同步发送与异步发送

1.什么是同步和异步

同步就是，串行，一条龙异步一起运行
举例：餐馆点餐
同步：需要等服务员过来，让服务员记录，
异步：点餐APP直接点餐，交给队列，让他自己运行

2.发送的同步异步

同步：需要得到返回值
异步：发送过去不管了

3. 分区好处

啥是分区？
将一个数据块分成多个数据块
将数据分布式处理了
存储：可以分在多个机器上，也可以整多个副本。便于存储，同时提高健壮性
IO：多个数据块可以同时进行发送接收消费。生产者可以以分区为单位发送数据，消费者可以以分区为单位进行消费

4. 默认分区器

前提条件： 1.分区 2.key值
规则：

1存在，按1分区
1不存在，按2.key值对分区数取余得到的值分区
1.2都不存在随机选个分区，等这个批次发送完了，再换

3 就是粘性分区
那么粘性分区的缺点是什么？
因为缓冲区溢出的条件是，大小和时间双重判断，如果大小不够，但是时间够了，还是会发走，这样，最后导致，分区上产生数据倾斜
如何解决的？
3.3.1 Kafka去掉粘性分区的时间控制，批次只由大小判断

1.3.自定义分区器

1.思路

1.实现接口Parititoner,重写相关方法
2.修改配置将partitioner设置为默认配置

2.1 自定义分区器代码

public class MyPartitioner implements Partitioner {//  自定义分区器 实现partitioner接口// 1.分区方法@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 获取消息String data = value.toString();// 创建partition 作为最后的分区标识int partitions;// 分区逻辑// 根据含有的字符串进行判断 判断进入哪个分区if (data.contains("atguigu")){partitions = 0;} else if (data.contains("shangguigu")){partitions = 1;} else {partitions = 2;}return partitions;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}

2.2 主类

package com.atguigu.producer;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;public class ProducerClientAsync {public static void main(String[] args) {// 0 配置对象Properties properties = new Properties();//  --指定kafka的Broker地址properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");//  -- 1.指定序列化器 序列化器的全限定类名properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//.setProperty(ProducerConfig.LINGER_MS_CONFIG,"0");// -- 2.设置分区器properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner.class.getName());// -- 3.获取客户端连接对象KafkaProducer<String,String> kafkaProducer= new KafkaProducer<String,String>(properties);//  key是主题  v是发送内容  这里注意一下// -- 4.发送数据String[] str= {"atguigu","111","atguigu","shangguigu","222"};for (int i =0; i < str.length; i++) {System.out.println(str[i]);try {kafkaProducer.send(new ProducerRecord<>("first", str[i]), new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception == null){System.out.println("主题：" + metadata.topic() + "->"  + "分区：" + metadata.partition());}else {// 出现异常打印exception.printStackTrace();}}}).get();} catch (InterruptedException e) {throw new RuntimeException(e);} catch (ExecutionException e) {throw new RuntimeException(e);}}kafkaProducer.close();}
}