文章目录

一、生产者消息发送流程
- 1. 发送流程
- 2. 生产者重要参数列表
二、异步发送 API
- 1. 普通异步发送
- 2. 带回调函数的异步发送
- 3. 同步发送
三、生产者分区
- 1. 分区好处
- 2. 生产者发送消息的分区策略
- - （1）默认的分区器 DefaultPartitioner
  - （2）案例一：将数据发往指定的 partition
  - （3）案例二：没有指明 partition 值，但有 key 的情况
- 3. 自定义分区器

一、生产者消息发送流程

1. 发送流程

在消息发送的过程中，涉及到了两个线程——main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

在这里插入图片描述

2. 生产者重要参数列表

参数名称	描述
bootstrap.servers	生产者连接集群所需的broker地址清单。例如：hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置1个或者多个，中间用逗号隔开。注意这里并非需要所有的broker地址，因为生产者可以从给定的broker里可以查找到其他broker信息。
key.serializer 和 value.serializer	指定发送消息的 key 和 value 的序列化类型。一定要写全类名。
buffer.memory	RecordAccumulator 缓冲区总大小，默认 32m。
batch.size	缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms	如果数据迟迟未达到batch.size，sender 等待 linger.time之后就会发送数据。单位ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据，Leader 收到数据后应答。-1（all）：生产者发送过来的数据，Leader和 isr 队列里面的所有节点收齐数据后应答。默认值是-1，-1 和all 是等价的。
max.in.flight.requests.per.connection	允许最多没有返回 ack 的次数，默认为 5，开启幂等性要保证该值是 1-5 的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是 100ms。
enable.idempotence	是否开启幂等性，默认 true，开启幂等性。
compression.type	生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。

二、异步发送 API

1. 普通异步发送

1.需求：创建 Kafka 生产者，采用异步的方式发送到 Kafka Broker
2.代码编写

（1）导入依赖

<dependencies>
	 <dependency>
		 <groupId>org.apache.kafka</groupId>
		 <artifactId>kafka-clients</artifactId>
		 <version>3.0.0</version>
	 </dependency>
</dependencies>

（2）编写不带回调函数的 API 代码

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;

public class producer {
    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用 send 方法，发送消息
        for (int i = 0; i < 5; i++) {    
            kafkaProducer.send(new ProducerRecord<>("first","kafka" + i));
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

3.测试：
（1）在 hadoop102 上开启 Kafka 消费者。
[root@hadoop102 kafka_2.12-3.0.0]# bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
（2）在 IDEA 中执行代码，观察 hadoop102 控制台中接收的消息。
kafka0
kafka1
kafka2
kafka3
kafka4

2. 带回调函数的异步发送

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
public class CustomProducerCallback {

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            // 添加回调
            kafkaProducer.send(new ProducerRecord<>("first", "kafka" + i), new Callback() {// 该方法在 Producer 收到 ack 时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition());
                    } else {
                        // 出现异常打印
                        exception.printStackTrace();
                    }
                }
            });
            // 延迟一会看到数据发往不同分区
            Thread.sleep(2);
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

1.观察 hadoop102 控制台中接收的消息。
kafka0
kafka1
kafka2
kafka3
kafka4
2.在 IDEA 控制台观察回调信息。
 主题： first->分区：0
 主题： first->分区：0
 主题： first->分区：0
 主题： first->分区：1
 主题： first->分区：1

3. 同步发送

只需在异步发送的基础上，再调用一下 get()方法即可。
kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();

三、生产者分区

1. 分区好处

（1）便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
（2）提高并行度，生产者以分区为单位发送数据；消费者以分区为单位进行消费数据。

在这里插入图片描述

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

（1）默认的分区器 DefaultPartitioner

DefaultPartitioner 源码注释：

/*The default partitioning strategy:（默认的分区策略:）
If a partition is specified in the record, use it（如果记录中指定了分区，则使用它）
If no partition is specified but a key is present choose a partition based on a hash of the key（如果没有指定分区但有一个key，则根据key的散列选择一个分区）
If no partition or key is present choose the sticky partition that changes when the batch is full. （如果不存在分区或key，则选择在批处理满时更改的粘性分区。）
See KIP-480 for details about sticky partitioning.（有关粘性分区的详细信息，请参见KIP-480。）
*/
public class DefaultPartitioner implements Partitioner {
	
}

ProducerRecord 构造方法

//（1）指明partition的情况下，直接将指明的值作为partition值；例如：partition=0，所有数据写入分区0。
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {}
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value){}
public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {}
public ProducerRecord(String topic, Integer partition, K key, V value) {}
//（2）没有指明partition值但有key的情况下，将key的hash值与topic的partition总数进行取余得到partition值；例如：key1的hash值=5， key2的hash值=6 ，topic的partition总数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。
public ProducerRecord(String topic, K key, V value) {}
//（3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到，Kafka再随机一个分区进行使用（如果还是0会继续随机）。
public ProducerRecord(String topic, V value) {}

（2）案例一：将数据发往指定的 partition

将数据发往指定 partition 的情况下，例如：将所有数据发往分区 1 中。

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) throws InterruptedException {

        // 0 配置
        Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        // 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // 1 创建kafka生产者对象
        // "" hello
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 2 发送数据
        for (int i = 0; i < 5; i++) {
        	// 指定数据发送到 1 号分区，key 为空（IDEA 中 ctrl + p 查看参数）
            kafkaProducer.send(new ProducerRecord<>("first", 1, "", "hello " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.println("主题： " + metadata.topic() + "->分区：" + metadata.partition());
                    }
                }
            });

            Thread.sleep(2);
        }

        // 3 关闭资源
        kafkaProducer.close();
    }
}

测试：

①在 hadoop102 上开启 Kafka 消费者。

[root@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

②在 IDEA 中执行代码，观察 hadoop102 控制台中接收的消息。

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
hello 0
hello 1
hello 2
hello 3
hello 4

③在 IDEA 控制台观察回调信息。

主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1

（3）案例二：没有指明 partition 值，但有 key 的情况

没有指明 partition 值，但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 总数进行取余得到 partition 值。

public class CustomProducerCallback {

    public static void main(String[] args) throws InterruptedException {

        // 0 配置
        Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");

        // 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        // 1 创建kafka生产者对象
        // "" hello
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 2 发送数据
        for (int i = 0; i < 500; i++) {
            // 依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，分别发往 1、2、0
            kafkaProducer.send(new ProducerRecord<>("first", "a", "hello " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null){
                        System.out.println("主题： "+metadata.topic() + "->分区："+ metadata.partition());
                    }
                }
            });

            Thread.sleep(2);
        }

        // 3 关闭资源
        kafkaProducer.close();
    }
}

测试：

①key="a"时，在控制台查看结果。

主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1
主题：first->分区：1

②key="b"时，在控制台查看结果。

主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2
主题：first->分区：2

③key="f"时，在控制台查看结果。

主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0

3. 自定义分区器

如果研发人员可以根据企业需求，自己重新实现分区器。

需求
实现一个分区器实现，发送过来的数据中如果包含 hello，就发往 0 号分区，不包含 hello，就发往 1 号分区。
实现步骤
（1）定义类实现 Partitioner 接口。
（2）重写 partition()方法。
（3）使用分区器的方法，在生产者的配置中添加分区器参数。

/**
 * 1. 实现接口 Partitioner
 * 2. 实现 3 个方法：partition,close,configure
 * 3. 编写 partition 方法，返回分区号
 */
public class MyPartitioner implements Partitioner {
    /**
     * 返回信息对应的分区
     * @param topic 主题
     * @param key 消息的 key
     * @param keyBytes 消息的 key 序列化后的字节数组
     * @param value 消息的 value
     * @param valueBytes 消息的 value 序列化后的字节数组
     * @param cluster 集群元数据可以查看分区信息
     * @return
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        // 获取数据 hello
        String msgValues = value.toString();

        // 创建 partition
        int partition;

        // 判断消息是否包含 hello
        if (msgValues.contains("hello")){
            partition = 0;
        }else {
            partition = 1;
        }

        // 返回分区号
        return partition;
    }

    // 关闭资源
    @Override
    public void close() {

    }

    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {

    }
}

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

public class CustomProducerCallbackPartitions {

    public static void main(String[] args) throws InterruptedException {

        // 0 配置
        Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        // 指定对应的key和value的序列化类型 key.serializer
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        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.producer.MyPartitioner");

        // 1 创建kafka生产者对象
        // "" hello
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 2 发送数据
        for (int i = 0; i < 5; i++) {
            // 指定数据发送到 1 号分区，key 为空（IDEA 中 ctrl + p 查看参数）
            kafkaProducer.send(new ProducerRecord<>("first", 1, "", "hello " + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.println("主题： " + metadata.topic() + "->分区： " + metadata.partition());
                    }
                }
            });

            Thread.sleep(2);
        }

        // 3 关闭资源
        kafkaProducer.close();
    }
}

测试：

①在 hadoop102 上开启 Kafka 消费者。

[root@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

②在 IDEA 控制台观察回调信息。

主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0
主题：first->分区：0

Kafka（四）- Kafka 生产者

文章目录

一、生产者消息发送流程

1. 发送流程

2. 生产者重要参数列表

二、异步发送 API

1. 普通异步发送

2. 带回调函数的异步发送

3. 同步发送

三、生产者分区

1. 分区好处

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

（1）默认的分区器 DefaultPartitioner

（2）案例一：将数据发往指定的 partition

（3）案例二：没有指明 partition 值，但有 key 的情况

3. 自定义分区器

相关文章

行业沙龙第四期丨企业供应链协同的数字化解痛之道

网络安全核心基础篇总结

基于51单片机水塔水箱液水位WIFI监控报警设计proteus仿真原理图PCB

快来给你的宠物视频加个表情特效吧

Android适配【入坑指南+解决痛点】

冷热电气多能互补的微能源网鲁棒优化调度附Matlab代码

基于Session的认证与授权实践

Matplotlib入门[05]——注释与标签

ag-Grid Enterprise v28.2.1 企业版注册版

（详解错误情况，及解决方法）Vue 数据更新了但页面没有更新的情况

你知道哪些常用快捷键？电脑快捷键大全，打工人必备！

演讲实录 | OpenMLDB 整合自动特征工程

RabbitMQ——RabbitMQ的六种工作模式详解

JVM之垃圾回收器一

对比分析小游戏引擎孰优孰劣

leetcode 638. 大礼包-思路整理

扩散模型(Diffusion)最新综述+GitHub论文汇总-A Survey On Generative Diffusion

Thread类及常见方法

世界杯征文活动 | 神奇！一段JavaScript代码生成会动的足球

python中的import详解