生产者_分区实战二

需求：不指定 partition 值但有 key的情况下，验证将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值。

1. 创建类NoPartitionHaveKey

2. 编写代码

   xxxxxxxxxx
   package com.itbaizhan.kafka.producer;
   
   import org.apache.kafka.clients.producer.*;
   import org.apache.kafka.common.serialization.StringSerializer;
   
   import java.util.Properties;
   import java.util.concurrent.ExecutionException;
   
   public class NoPartitionHaveKey {
       public static void main(String[] args) throws ExecutionException, InterruptedException {
           //TODO 1.声明并实例化Kafka Producer的配置文件对象
           Properties prop = new Properties();
           //TODO 2.为配置文件对象设置参数
           // 2.1 配置bootstrap_servers
           prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node2:9092,node3:9092,node4:9092");
           // 2.2 配置key和value的序列化类
           prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
           prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
           //TODO 3.声明并实例化生产者对象
           KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
           //TODO 4.发送消息
           for(int i = 0;i<5;i++){
            //未指定分区号，key为"gtjin"测试运行一次，改为"kafka"后再测试一次。
               producer.send(new ProducerRecord<>("topicA","gtjin", "unsync_msg" + i),
                   new Callback() {
                       //如下方法在生产者收到acks确认时异步调用
                       @Override
                       public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                           if(e == null){
                               //无异常信息，输出主题和分区信息到控制台
                               System.out.println("topic:"+recordMetadata.topic()
                                       +",partition:"+recordMetadata.partition());
                           }else{//打印异常信息
                               System.out.println(e.getMessage());
                           }
                       }
                   });
               Thread.sleep(5);
           }
           //TODO 5.关闭生产者
           producer.close();
       }
   }
   

3. 在node2上开启Kafka消费者进行消费

   xxxxxxxxxx
   [root@node2 ~]# kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topicA
   

4. 运行ProducerToPartition类

5. 观察node2上Kafka消费者消费消息的情况

   xxxxxxxxxx
   [root@node2 ~]# kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topicA
   unsync_msg0
   unsync_msg1
   unsync_msg2
   unsync_msg3
   unsync_msg4
   

6. 控制台输出信息

   xxxxxxxxxx
   topic:topicA,partition:1
   topic:topicA,partition:1
   topic:topicA,partition:1
   topic:topicA,partition:1
   topic:topicA,partition:1
   

7. 修改ProducerToPartition类，将key改为"kafka",再次运行。

8. 观察node2上Kafka消费者消费消息的情况

   xxxxxxxxxx
   unsync_msg0
   unsync_msg1
   unsync_msg2
   unsync_msg3
   unsync_msg4
   

9. 控制台输出信息

   xxxxxxxxxx
   topic:topicA,partition:0
   topic:topicA,partition:0
   topic:topicA,partition:0
   topic:topicA,partition:0
   topic:topicA,partition:0