一、生产者分区优点

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

二、生产者发送消息的分区策略

2.1、默认的分区器

• 在 IDEA中 ctrl +n，全局查找 DefaultPartitioner类，DefaultPartitioner类实现了Partitioner接口。
  

2.2、指定分区(partition)值

• 在IDEA 中全局查找 （ctrl +n ）ProducerRecord类 ， 在类中可以看到如下4个构造方法：
• 如下4个构造方法：指明了partition 的情况下，直接将指明的值作为partition值 ；
• 例如partition=0，所有数据写入分区0
  

2.3、没有指明分区(partition)值，但有key 的情况

• 在IDEA 中全局查找 （ctrl +n ）ProducerRecord类 ， 在类中可以看到如下1个构造方法：
• 没有指明partition 值但有key 的情况下，将key 的hash值与topic的partition数进行 取余 得到partition 值；
• 例如：key1 的hash 值=5，key2 的hash 值=6 ，topic 的partition 数=2，那么 么key1 对应的value1 写入1 号分区，key2 对应的value2 写入0
  

2.4、 既没有分区(partition)值，又没有key 值的情况

• 在IDEA 中全局查找 （ctrl +n ）ProducerRecord类 ， 在类中可以看到如下1个构造方法：

• Kafka 采用Sticky Partition（ 黏性分区器 ） ， 会随机选择一个分区 ， 并尽可能一直使用该分区 ， 待该分区的batch已满或者已完成，Kafka 再随机一个分区进行使用（ 和上一次的分区不同） 。

• 例如：第一次随机选择0 号分区 ，等0 号分区当前批次满了 （ 默认16k）或者linger.ms 设置的时间到，Kafka 再随机一个分区进行使用 （ 如果还是0 会继续随机 ）

  

三、指定分区(partition)值的代码示例

• 指定数据发送到 0 号分区的代码示例

  package com.xz.kafka.producer;
  
  import org.apache.kafka.clients.producer.*;
  import org.apache.kafka.common.serialization.StringSerializer;
  
  import java.util.Properties;
  
  /**
   * @author: xz
   * @since: 2023/4/2 22:02
   * @description: 生产者将数据发往指定 partition的情况
   */
  public class CustomProducerCallbackConfirmPartitions {
      public static void main(String[] args) throws InterruptedException {
  
          //1、创建 kafka 生产者的配置对象
          Properties properties = new Properties();
  
          //2、给 kafka 配置对象添加配置信息：bootstrap.servers
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
  
          //3、指定对应的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());
  
          //4、创建 kafka 生产者对象
          KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
  
          //5、调用 send 方法,发送消息
          for (int i = 0; i < 5; i++) {
              //指定数据发送到 0 号分区，key 为空
              kafkaProducer.send(new ProducerRecord<>("news", 0,"","hello kafka" + i), new Callback() {
                  @Override
                  public void onCompletion(RecordMetadata metadata, Exception exception) {
                      if (exception == null){
                          System.out.println("主题： "+metadata.topic() + " 分区： "+ metadata.partition());
                      }else {
                          exception.printStackTrace();
                      }
                  }
              });
              Thread.sleep(2);
          }
  
          // 3 关闭资源
          kafkaProducer.close();
      }
  }
  

• 在kafka集群某一台服务器上开启 Kafka 消费者

  [root@localhost kafka-3.0.0]# bin/kafka-console-consumer.sh  --bootstrap-server 192.168.136.27:9092 --topic news
  

  

• 在 IDEA 中执行代码，观察 开启 Kafka 消费者的服务器中是否接收到消息。如下图所示：
  

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

四、没有指明分区(partition)值，但有key值的情况代码示例

• 代码示例

  package com.xz.kafka.producer;
  
  import org.apache.kafka.clients.producer.*;
  import org.apache.kafka.common.serialization.StringSerializer;
  import java.util.Properties;
  
  /**
   * @author: xz
   * @since: 2023/4/2 22:25
   * @description: 生产者发送数据没有指明 partition 值，但有key值的情况下
   */
  public class CustomProducerCallbackNoPartitionsConfirmKey {
  
      public static void main(String[] args) throws InterruptedException {
  
          //1、创建 kafka 生产者的配置对象
          Properties properties = new Properties();
  
          //2、给 kafka 配置对象添加配置信息：bootstrap.servers
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
  
          //3、指定对应的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());
  
          //4、创建 kafka 生产者对象
          KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
  
          //5、调用 send 方法,发送消息
          for (int i = 0; i < 3; i++) {
              //依次指定 key 值为 a,b,f ，数据 key 的 hash 值与 3 个分区求余，分别发往 1、2、0
              kafkaProducer.send(new ProducerRecord<>("news", "a","hello kafka" + i), new Callback() {
                  @Override
                  public void onCompletion(RecordMetadata metadata, Exception exception) {
                      if (exception == null){
                          System.out.println("主题： "+metadata.topic() + " 分区： "+ metadata.partition());
                      }else {
                          exception.printStackTrace();
                      }
                  }
              });
              Thread.sleep(2);
          }
  
          // 3 关闭资源
          kafkaProducer.close();
      }
  }
  

• 在kafka集群某一台服务器上开启 Kafka 消费者

  [root@localhost kafka-3.0.0]# bin/kafka-console-consumer.sh  --bootstrap-server 192.168.136.27:9092 --topic news
  

  

• 当key="a"时，在控制台查看结果。如下图：
  

• 当key="b"时，在控制台查看结果。如下图：
  

• 当key="f"时，在控制台查看结果。如下图：
  

五、 既没有分区(partition)值，又没有key 值的情况代码示例

• 代码示例

  package com.xz.kafka.producer;
  
  import org.apache.kafka.clients.producer.*;
  import org.apache.kafka.common.serialization.StringSerializer;
  
  import java.util.Properties;
  
  /**
   * @author: xz
   * @since: 2023/4/2 22:49
   * @description: 生产者发送数据 既没有分区(partition)值，又没有key值
   */
  public class CustomProducerCallbackNoPartitionsAndKey {
      public static void main(String[] args) throws InterruptedException {
  
          //1、创建 kafka 生产者的配置对象
          Properties properties = new Properties();
  
          //2、给 kafka 配置对象添加配置信息：bootstrap.servers
          properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.27:9092,192.168.136.28:9092,192.168.136.29:9092");
  
          //3、指定对应的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());
  
          //4、创建 kafka 生产者对象
          KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
  
          //5、调用 send 方法,发送消息
          for (int i = 0; i < 5; i++) {
              kafkaProducer.send(new ProducerRecord<>("news", "hello kafka" + 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();
      }
  }
  

• 在kafka集群某一台服务器上开启 Kafka 消费者

  [root@localhost kafka-3.0.0]# bin/kafka-console-consumer.sh  --bootstrap-server 192.168.136.27:9092 --topic news
  

  

• 在 IDEA 中执行代码，观察开启 Kafka 消费者的服务器中是否接收到消息。如下图所示：
  

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