文章目录

• 一、生产者消息发送流程
• • 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. 生产者重要参数列表

二、异步发送 API

1. 普通异步发送

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

（1）导入依赖

org.apache.kafkakafka-clients3.0.0

（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.serversproperties.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 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.serializerproperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 3. 创建 kafka 生产者对象KafkaProducer kafkaProducer = new KafkaProducer<>(properties);// 4. 调用 send 方法,发送消息for (int i = 0; i < 5; i++) {// 添加回调kafkaProducer.send(new ProducerRecord<>("first", "kafka" + i), new Callback() {// 该方法在 Producer 收到 ack 时调用，为异步调用@Overridepublic 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
 headers) {}
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value){}
public ProducerRecord(String topic, Integer partition, K key, V value, Iterable
 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.serversproperties.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生产者对象// "" helloKafkaProducer 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() {@Overridepublic 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.serversproperties.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生产者对象// "" helloKafkaProducer kafkaProducer = new KafkaProducer<>(properties);// 2 发送数据for (int i = 0; i < 500; i++) {// 依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，分别发往 1、2、0kafkaProducer.send(new ProducerRecord<>("first", "a", "hello " + i), new Callback() {@Overridepublic 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. 自定义分区器

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

1. 需求
   实现一个分区器实现，发送过来的数据中如果包含 hello，就发往 0 号分区，不包含 hello，就发往 1 号分区。
2. 实现步骤
   （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*/@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 获取数据 helloString msgValues = value.toString();// 创建 partitionint partition;// 判断消息是否包含 helloif (msgValues.contains("hello")){partition = 0;}else {partition = 1;}// 返回分区号return partition;}// 关闭资源@Overridepublic void close() {}// 配置方法@Overridepublic void configure(Map configs) {}
}

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

public class CustomProducerCallbackPartitions {public static void main(String[] args) throws InterruptedException {// 0 配置Properties properties = new Properties();// 连接集群 bootstrap.serversproperties.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生产者对象// "" helloKafkaProducer 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() {@Overridepublic 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