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

news2025/2/9 6:50:31

生产者学习

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.分区方法
    @Override
    public 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;
    }

    @Override
    public void close() {

    }

    @Override
    public 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() {
                    @Override
                    public 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();
    }
}