Kafka Java API+自定义分区

2021/6/2 20:22:44

本文主要是介绍Kafka Java API+自定义分区,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

kafka的API

第一步:导入kafka的开发jar包

 


   

   

 


 

 

Kafka生产者


@Test

   public void kafkaProducer() throws Exception {

      //1、准备配置文件

       Properties props = new Properties();

       props.put("bootstrap.servers", "hadoop-001:9092,hadoop-002:9092,hadoop-003:9092");

       props.put("acks", "all");

       props.put("retries", 0);

       props.put("batch.size", 16384);

       props.put("linger.ms", 1);

       props.put("buffer.memory", 33554432);

       props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

       props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

       //2、创建KafkaProducer

       KafkaProducer

       for (int i=0;i<100;i++){

           //3、发送数据

           kafkaProducer.send(new ProducerRecord

       }

 

      kafkaProducer.close();

   }


 

 

Kafka消费者


@Test

   public void kafkaConsum() throws Exception {

        // 1、准备配置文件

       Properties props = new Properties();

       props.put("bootstrap.servers", "hadoop-001:9092,hadoop-002:9092,hadoop-003:9092");

       props.put("group.id", "test");

       props.put("enable.auto.commit", "true");

       props.put("auto.commit.interval.ms", "1000");

       props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

       props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

 

       // 2、创建KafkaConsumer

       KafkaConsumer

       // 3、订阅数据,这里的topic可以是多个

       kafkaConsumer.subscribe(Arrays.asList("yun01"));

       // 4、获取数据

       while (true) {

           ConsumerRecords

           for (ConsumerRecord

               System.out.printf("topic = %s,offset = %d, key = %s, value = %s%n",record.topic(), record.offset(), record.key(), record.value());

           }

 

       }

   }

 


 

 

 

kafka的自定义分区

第一种方式:直接指定分区


kafkaProducer.send(new ProducerRecord

 

第二种自定义分区

public class KafkaCustomPartitioner implements Partitioner {

   @Override

   public void configure(Map

   }

 

   @Override

   public int partition(String topic, Object arg1, byte[] keyBytes, Object arg3, byte[] arg4, Cluster cluster) {

      List

       int partitionNum = partitions.size();

      Random random = new Random();

      int partition = random.nextInt(partitionNum);

       return partition;

   }

 

   @Override

   public void close() {

     

   }

 

}


 

 

 

主代码中添加配置

 


@Test

   public void kafkaProducer() throws Exception {

      //1、准备配置文件

       Properties props = new Properties();

       props.put("bootstrap.servers", "hadoop-001:9092,hadoop-002:9092,hadoop-003:9092");

       props.put("acks", "all");

       props.put("retries", 0);

       props.put("batch.size", 16384);

       props.put("linger.ms", 1);

       props.put("buffer.memory", 33554432);

       props.put("partitioner.class", "com.gec.kafkaclient.MyCustomerPartitons");

       props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

       props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

       //2、创建KafkaProducer

       KafkaProducer

       for (int i=0;i<100;i++){

           //3、发送数据

           kafkaProducer.send(new ProducerRecord

       }

 

      kafkaProducer.close();

   }


 



这篇关于Kafka Java API+自定义分区的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程