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）。

重复数据的判断标准：具有<PID, Partition, SeqNumber>相同主键的消息提交时，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();
        }
    }
}