devbox
神奇cpp
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Kafka快速入门：Kafka驱动JavaApi的使用

作者：神奇cpp | 2024-08-22 20:14:35

赞

踩

Kafka快速入门：Kafka驱动JavaApi的使用

生产者和消费者是Kafka的核心概念之一，它们在客户端被创建和使用，并且包含了许多与Kafka性能和机制相关的配置。虽然Kafka提供的命令行工具能够执行许多基本操作，但它无法实现所有可能的性能优化。相比之下，使用Java API可以充分利用编程语言的灵活性，对生产者和消费者进行更精细的性能调优。对于大多数中间件，熟悉服务器的命令行操作可能足以帮助学习其API的使用。然而，Kafka则不同，要全面掌握Kafka的所有特性，必须系统地学习和理解其Java API。

1. 基础消费流程

在javaApi中可以通过创建一个Kafka生产者和消费者的配置对象，在new生产者或消费者的类时将配置对象传入，然后生产者实例通过调用send方法发送数据，消费者通过poll方法消费数据，数据需要通过ProducerRecords类封装key和value，并在生产者和消费者配置中为key和value指定序列化和反序列化类（key可以传null，key是在日志回收策略中发挥作用）。经过这样一套操作，消息就可以成功从生产者发往消费者。

  1. package com.kafak.testkafka;
  2.  
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.clients.producer.KafkaProducer;
  8. import org.apache.kafka.clients.producer.ProducerConfig;
  9. import org.apache.kafka.clients.producer.ProducerRecord;
  10. import org.junit.jupiter.api.Test;
  11. import org.springframework.boot.test.context.SpringBootTest;
  12.  
  13. import java.time.Duration;
  14. import java.util.Properties;
  15.  
  16. @SpringBootTest
  17. class TestKafkaApplicationTests {
  18.  
  19.     //预定义Kafka对象实例，因为Kafka对象时线程安全，所以可以定义外面节省资源防止重复创建
  20.     KafkaProducer<String, String> kafkaProducer;
  21.  
  22.     //创建生产者
  23.     public KafkaProducer<String, String> getKafkaProducer() {
  24.  
  25.         //创建生产者配置
  26.         Properties props = new Properties();
  27.         //配置Kafka集群地址
  28.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  29.         //配置序列化
  30.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  31.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  32.         //返回生产者
  33.         return new KafkaProducer<String, String>(props);
  34.     }
  35.  
  36.     //创建消费者
  37.     public KafkaConsumer<String, String> getKafkaConsumer() {
  38.         //创建消费者配置
  39.         Properties props = new Properties();
  40.         //配置Kafka集群地址
  41.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  42.         //配置消费者组id
  43.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
  44.         //配置反序列化
  45.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  46.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  47.         //返回消费者
  48.         return new KafkaConsumer<>(props);
  49.     }
  50.  
  51.     //通过生产者生产一百条数据
  52.     @Test
  53.     void kafkaProducerTest() {
  54.         //获取生产者
  55.         kafkaProducer = getKafkaProducer();
  56.         //发送消息
  57.         for (int i = 0; i < 100; i++) {
  58.             kafkaProducer.send(new ProducerRecord<String, String>("topicJava", "testKey" + i, "testValue" + i));
  59.         }
  60.     }
  61.  
  62.     //通过消费者消费消息
  63.     @Test
  64.     void kafkaConsumerTest() {
  65.         //创建消费者，由于消费者是线程不安全，所以使用一次实例化一次，可以方式出现线程安全问题
  66.         KafkaConsumer<String, String> kafkaConsumer = getKafkaConsumer();
  67.         //接受消费者信息，传入100毫秒，消费者会一百毫秒拉去一次消息
  68.         ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
  69.         // 处理消息
  70.         for (ConsumerRecord<String, String> record : records) {
  71.             System.out.printf("Consumed message with key %s, value %s, from partition %d with offset %d%n",
  72.                     record.key(), record.value(), record.partition(), record.offset());
  73.         }
  74.     }
  75.  
  76. }

2. 消息确认

消息确认的原理性知识可以通过下面这篇文章学习，这里主要讲实操。

Kafka运行机制（二）：消息确认，消息日志的存储和回收icon-default.png?t=N7T8https://blog.csdn.net/dxh9231028/article/details/141329851?spm=1001.2014.3001.5501

生产者端

生产者端的消息确认策略由acks配置项控制，其由三种配置方式，其中我在下面这篇文章中详细讲述了相关知识。我们可以通过javaApi配置acks来控制确认策略。

生产者端的消息确认有同步和异步两种方式。

  • 同步消息确认：同步消息确认是生产者实例在调用send方法后紧接着调用get方法，该方法会阻塞线程的继续执行，等待消息发送结果。当消息发送失败时，如果配置了重试机制（通过设置 retries 属性），生产者会自动重试指定的次数。如果在所有重试尝试后仍然失败，最终会抛出异常，通知调用方消息发送失败。
  • 异步消息确认：异步消息确认通过生产者实例调用send方法时传入，第二个回调函数的参数。在成功或失败响应时，执行回调函数，异步消息确认不会阻塞代码，也不会触发自动重试。

消费者端

消费者消费成功在客户端的体现是成功获取到了数据，这本没有什么好说的，不过消费者不仅需要响应客户端数据，还要讲偏移量发送给Kafka，在这一过程中，消费者提供了手动提交和自动提交两种方式。启动自动提交是默认开启的，而手动提交则需要配置enable.auto.commit为false，然后通过创建分区和偏移量的映射关系，通过消费者的commit方法提交偏移量。

代码实现

下面代码中，我创建了四个测试单元，其中前两个测试单元，分别是生产者同步提交和异步提交，而后两个测试单元分别时消费者的自动提交和手动提交。

  1. package com.kafak.testkafka;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.clients.producer.RecordMetadata;
  8. import org.apache.kafka.common.TopicPartition;
  9. import org.junit.jupiter.api.Test;
  10. import org.springframework.boot.test.context.SpringBootTest;
  11.  
  12. import java.time.Duration;
  13. import java.util.HashMap;
  14. import java.util.Map;
  15. import java.util.Properties;
  16.  
  17. @SpringBootTest
  18. class TestKafkaApplicationTests {
  19.     //预定义Kafka对象实例，因为Kafka对象时线程安全，所以可以定义外面节省资源防止重复创建
  20.     KafkaProducer<String, String> kafkaProducer;
  21.  
  22.     //创建生产者
  23.     public KafkaProducer<String, String> getKafkaProducer() {
  24.         //创建生产者配置
  25.         Properties props = new Properties();
  26.  
  27.         //配置消息确认策略
  28.         props.put(ProducerConfig.ACKS_CONFIG, "all");
  29.  
  30.         //配置重试次数
  31.         props.put(ProducerConfig.RETRIES_CONFIG,3);
  32.  
  33.         //配置Kafka集群地址
  34.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  35.         //配置序列化
  36.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  37.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  38.         //返回生产者
  39.         return new KafkaProducer<String, String>(props);
  40.     }
  41.  
  42.     //创建消费者
  43.     public KafkaConsumer<String, String> getKafkaConsumer(Boolean isAutoCommit) {
  44.         //创建消费者配置
  45.         Properties props = new Properties();
  46.         //配置Kafka集群地址
  47.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  48.         //配置消费者组id
  49.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
  50.         //配置反序列化
  51.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  52.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  53.         //判断当前消费者是否开启自动提交
  54.         if (!isAutoCommit) {
  55.             //关闭自动提交
  56.             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  57.         }else{
  58.             //设置自动提交间隔时间1s
  59.             props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
  60.         }
  61.         //返回消费者
  62.         return new KafkaConsumer<>(props);
  63.     }
  64.  
  65.     //生产者同步确认
  66.     @Test
  67.     void kafkaProducerGetTest() {
  68.         if(kafkaProducer == null) {
  69.             kafkaProducer = getKafkaProducer();
  70.         }
  71.         //同步确认消息是否发送成功
  72.         for (int i = 0; i < 100; i++) {
  73.             try{
  74.                 RecordMetadata topicJava = kafkaProducer.send(new ProducerRecord<String, String>("topicJava", "testKey" + i, "testValue" + i)).get();
  75.             }catch (Exception e) {
  76.                 e.printStackTrace();
  77.             }
  78.         }
  79.         kafkaProducer.close();
  80.     }
  81.     
  82.     //生产者异步确认
  83.     @Test
  84.     void kafkaProduceSyncTest() {
  85.         if(kafkaProducer == null) {
  86.             kafkaProducer = getKafkaProducer();
  87.         }
  88.         //异步确认是否发送成功
  89.         for (int i = 0; i < 100; i++) {
  90.             kafkaProducer.send(new ProducerRecord<String, String>("topicJava", "testKey" + i, "testValue" + i), (metadata, exception) -> {
  91.                 if (exception == null) {
  92.                     System.out.printf("发送消息成功, metadata=%s%n", metadata);
  93.                 } else {
  94.                     System.err.printf("发送消息失败, exception=%s%n", exception.getMessage());
  95.                 }
  96.             });
  97.         }
  98.         kafkaProducer.close();
  99.     }
  100.  
  101.     //消费者自动提交
  102.     @Test
  103.     void kafkaAutoCommitConsumerTest() {
  104.         //创建消费者开启自动提交
  105.         KafkaConsumer<String, String> kafkaConsumer = getKafkaConsumer(true);
  106.         //消费数据流程中无需负责偏移量提交
  107.         while (true) {
  108.             //接受消费者信息，传入100毫秒，消费者会一百毫秒拉去一次消息
  109.             ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
  110.             //处理消息
  111.             for (ConsumerRecord<String, String> record : records) {
  112.                 System.out.printf("消息消费成功, key=%s, value=%s, partition=%d, offset=%d%n",
  113.                         record.key(), record.value(), record.partition(), record.offset());
  114.             }
  115.         }
  116.     }
  117.  
  118.     //消费者手动提交
  119.     @Test
  120.     void kafkaSyncCommitConsumerTest() {
  121.         //创建消费者关闭自动提交
  122.         KafkaConsumer<String, String> kafkaConsumer = getKafkaConsumer(false);
  123.         //消费数据流程中需要在消费数据后，提交偏移量
  124.         while (true) {
  125.             //接受消费者信息，传入100毫秒，消费者会一百毫秒拉去一次消息
  126.             ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
  127.             // 处理消息
  128.             for (ConsumerRecord<String, String> record : records) {
  129.                 System.out.printf("消息消费成功, key=%s, value=%s, partition=%d, offset=%d%n",
  130.                         record.key(), record.value(), record.partition(), record.offset());
  131.                 //创建分区和偏移量的映射类
  132.                 Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
  133.                 //讲分区和偏移量的数据存入映射类
  134.                 offsets.put(new TopicPartition(record.topic(), record.partition()),
  135.                         new OffsetAndMetadata(record.offset() + 1));
  136.                 //偏移量提交
  137.                 kafkaConsumer.commitSync(offsets);
  138.             }
  139.         }
  140.     }
  141.  
  142. }

3. 批处理

批处理在生产者端，和消费者端也有不同的实现。我在Kakfa基本概念一文中清楚的讲解了批处理的概念，文章如下

Kafka基本概念icon-default.png?t=N7T8https://blog.csdn.net/dxh9231028/article/details/141270920?spm=1001.2014.3001.5501

生产者

在生产者端，生产者实例的send方法会发送消息到缓冲区中，而缓冲区消息何时发送给Kafka集群，则是通过配置batch.size和linger.ms配置，来实现当缓冲区存入多少消息，和距离上一次发送消息多久后，来发送这一轮缓冲区的消息到Kafka集群，代码实现如下

  1.     //创建生产者
  2.     public KafkaProducer<String, String> getKafkaProducer() {
  3.         //创建生产者配置
  4.         Properties props = new Properties();
  5.  
  6.         //配置生产者批处理
  7.         //缓冲区大小最大为16384比特
  8.         props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  9.         //距离上次发送消息时间隔3s
  10.         props.put(ProducerConfig.LINGER_MS_CONFIG,"3000");
  11.         
  12.         //配置消息确认策略
  13.         props.put(ProducerConfig.ACKS_CONFIG, "all");
  14.         //配置重试次数
  15.         props.put(ProducerConfig.RETRIES_CONFIG,3);
  16.         //配置Kafka集群地址
  17.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  18.         //配置序列化
  19.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  20.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  21.         //返回生产者
  22.         return new KafkaProducer<String, String>(props);
  23.     }

消费者

消费者端批处理，消费者在拉去消息时，会在fetch.max.bytes，max.partition.fetch.bytes和max.poll.records三个配置项，以及传入poll方法的超时时间参数的限制下，尽可能多的拉取更多消息。

  • fetch.max.bytes：fetch.max.bytes 是指 Kafka 消费者单次从服务器拉取数据时能够获取的最大字节数。这是全局的上限，控制每次 poll() 操作可以拉取的数据量总和。 默认值：50MB（即 52428800 字节）。
  • max.partition.fetch.bytes：max.partition.fetch.bytes 是指 Kafka 消费者从单个分区拉取消息时能获取的最大字节数，当消费者数量少于主题分区数量时，一个消费者可能会负责多个分区。默认值：1MB（即 1048576 字节）。
  • max.poll.records：max.poll.records 是指 Kafka 消费者每次调用 poll() 方法时能够拉取的最大消息条数。 默认值：500 条消息。

代码实现如下

  1.     //创建消费者
  2.     public KafkaConsumer<String, String> getKafkaConsumer(Boolean isAutoCommit) {
  3.         //创建消费者配置
  4.         Properties props = new Properties();
  5.  
  6.         //消费者批处理相关配置
  7.         //消费缓冲区大小，也就是一次消费最多能消费多少比特消息
  8.         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,16384);
  9.         //一次消费一个分区最多能消费多少比特
  10.         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,8192);
  11.         //一次消费最多能消费多少条数据
  12.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,1000);
  13.  
  14.         //配置Kafka集群地址
  15.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  16.         //配置消费者组id
  17.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
  18.         //配置反序列化
  19.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  20.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  21.         //判断当前消费者是否开启自动提交
  22.         if (!isAutoCommit) {
  23.             //关闭自动提交
  24.             props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  25.         }else{
  26.             //设置自动提交间隔时间1s
  27.             props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
  28.         }
  29.         //返回消费者
  30.         return new KafkaConsumer<>(props);
  31.     }

4. 事务操作

Kafka驱动支持事务操作，允许许生产者在多个主题和分区上以原子方式写入消息。这意味着你可以确保一组消息要么全部成功写入Kafka，要么全部失败。

事务操作首先通过生产者实例调用生产者实例的initTransactions方法，向kafka集群申请一个映射当前生产者的事务Id，然后就可以通过调用生产者实例的beginTransaction方法，开启一个事务，进行消息发送，最终通过调用commitTransaction方法完成事务的提交，如果中途发生异常则通过abortTransaction对当前事务进行回滚，代码实例如下

  1. package com.kafak.testkafka;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.clients.producer.KafkaProducer;
  5. import org.apache.kafka.clients.producer.ProducerConfig;
  6. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import org.apache.kafka.clients.producer.RecordMetadata;
  8. import org.apache.kafka.common.TopicPartition;
  9. import org.junit.jupiter.api.Test;
  10. import org.springframework.boot.test.context.SpringBootTest;
  11.  
  12. import java.time.Duration;
  13. import java.util.HashMap;
  14. import java.util.Map;
  15. import java.util.Properties;
  16.  
  17. @SpringBootTest
  18. class TestKafkaApplicationTests {
  19.     //预定义Kafka对象实例，因为Kafka对象时线程安全，所以可以定义外面节省资源防止重复创建
  20.     KafkaProducer<String, String> kafkaProducer;
  21.  
  22.     //创建生产者
  23.     public KafkaProducer<String, String> getKafkaProducer() {
  24.         //创建生产者配置
  25.         Properties props = new Properties();
  26.         //配置生产者批处理
  27.         //缓冲区大小最大为16384比特
  28.         props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  29.         //距离上次发送消息时间隔3s
  30.         props.put(ProducerConfig.LINGER_MS_CONFIG,"3000");
  31.         //配置消息确认策略
  32.         props.put(ProducerConfig.ACKS_CONFIG, "all");
  33.         //配置重试次数
  34.         props.put(ProducerConfig.RETRIES_CONFIG,3);
  35.         //配置Kafka集群地址
  36.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  37.         //配置序列化
  38.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  39.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  40.         //返回生产者
  41.         return new KafkaProducer<String, String>(props);
  42.     }
  43.  
  44.  
  45.  
  46.     //测试事务
  47.     @Test
  48.     void kafkaProducerTransactionTest() {
  49.         if(kafkaProducer == null) {
  50.             kafkaProducer = getKafkaProducer();
  51.         }
  52.         kafkaProducer.initTransactions();
  53.         try{
  54.             kafkaProducer.beginTransaction();
  55.             //消息发送相关操作
  56.             for (int i = 0; i < 100; i++) {
  57.                 try{
  58.                     RecordMetadata topicJava = kafkaProducer.send(new ProducerRecord<String, String>("topicJava", "testKey" + i, "testValue" + i)).get();
  59.                 }catch (Exception e) {
  60.                     e.printStackTrace();
  61.                 }
  62.             }
  63.             kafkaProducer.commitTransaction();
  64.         }catch (Exception e) {
  65.             e.printStackTrace();
  66.             kafkaProducer.abortTransaction();
  67.         }
  68.         kafkaProducer.close();
  69.     }
  70.  
  71.     
  72.  
  73. }

5. 自定义分区器

Kafka允许用户自定义分区器，实现特定的分区策略。可以通过实现Partitioner接口来创建自定义分区器。实现Partitioner接口需要实现三个方法，分别是partition，configure，close。

partition方法

partition方法是实现分区逻辑其的主要方法，其接受六个参数，分别是

  • String topic：消息要发送到的Kafka主题名称
  • Object key：消息的 key，可能为 null。
  • byte[] keyBytes：序列化后的 key，可能为 null
  • Object value：消息的 value，可以为任意对象
  • byte[] valueBytes：序列化后的 value，可能为 null。
  • Cluster cluster：Kafka集群的元数据信息，包括主题的分区数、每个分区的领导者等

partition方法的返回值则是发送分区的编号，通过这个机制可以实现不同逻辑的分区器。

configure方法

configuer方法在自定义分区类初始化时调用，当设计一些复杂操作，比如在发送消息前要和数据库交互时，可以在configure中完成数据库的连接。

close方法

close在分区逻辑执行完后调用，和configure一样，在复杂操作时，用于关闭分区逻辑中创建的连接，或一些内存资源等

假设我有一个三主机集群，其中30主机性能最好，31其次，32最差，我要通过自定义分区，将消息发送到三个分区的比例为3：2：1，通过Partitioner接口，可以简单的通过如下方式实现

  1. package com.kafak.testkafka;
  2.  
  3. import org.apache.kafka.clients.producer.Partitioner;
  4. import org.apache.kafka.common.Cluster;
  5. import org.apache.kafka.common.Node;
  6. import org.apache.kafka.common.PartitionInfo;
  7.  
  8. import java.util.List;
  9. import java.util.Map;
  10.  
  11. public class CustomPartitioner implements Partitioner {
  12.  
  13.     @Override
  14.     public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
  15.         //获取分区元数据
  16.         List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(s);
  17.         //创建一个0-100的随机数
  18.         double num = Math.random() * 100;
  19.         //默认传递分区号
  20.         Integer finalPartition = 0;
  21.         for (PartitionInfo partitionInfo : partitionInfos) {
  22.             //获取分区的leader
  23.             Node leader = partitionInfo.leader();
  24.             //获取分区leader的ip和端口
  25.             String leaderAddress = leader.host() + ":" + leader.port(); // 生成 "host:port" 格式的字符串
  26.             //如果随机数在0-50之间，发送消息至192.168.142.30:9092
  27.             if (num < 50 && leaderAddress.equals("192.168.142.30:9092")) {
  28.                 finalPartition = partitionInfo.partition();
  29.                 break; 
  30.             //如果随机数在50-82之间，发送消息至192.168.142.31:9092
  31.             } else if (num < 82 && num >= 50 && leaderAddress.equals("192.168.142.31:9092")) {
  32.                 finalPartition = partitionInfo.partition();
  33.                 break;
  34.             //如果随机数在82-100之间，发送消息至192.168.142.32:9092
  35.             } else if (num < 100 && num >= 82 && leaderAddress.equals("192.168.142.32:9092")) {
  36.                 finalPartition = partitionInfo.partition();
  37.                 break;
  38.             }
  39.         }
  40.         //返回最终分区号
  41.         return finalPartition;
  42.     }
  43.  
  44.  
  45.     @Override
  46.     public void close() {
  47.  
  48.     }
  49.     
  50.     @Override
  51.     public void configure(Map<String, ?> map) {
  52.  
  53.     }
  54. }

在生产者配置中通过partitioner_class配置自定义分区器，代码如下

  1.  public KafkaProducer<String, String> getKafkaProducer() {
  2.         //创建生产者配置
  3.         Properties props = new Properties();
  4.         
  5.         //启用自定义分区器
  6.         props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.kafak.testkafka.CustomPartitioner");
  7.         
  8.         //配置生产者批处理
  9.         //缓冲区大小最大为16384比特
  10.         props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  11.         //距离上次发送消息时间隔3s
  12.         props.put(ProducerConfig.LINGER_MS_CONFIG,"3000");
  13.         //配置消息确认策略
  14.         props.put(ProducerConfig.ACKS_CONFIG, "all");
  15.         //配置重试次数
  16.         props.put(ProducerConfig.RETRIES_CONFIG,3);
  17.         //配置Kafka集群地址
  18.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.30:9092,192.168.142.31:9092,192.168.142.32:9092");
  19.         //配置序列化
  20.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  21.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  22.         //返回生产者
  23.         return new KafkaProducer<String, String>(props);
  24.     }

如此，便可以实现一个自定义分区策略。 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/神奇cpp/article/detail/1017823
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号