Kafa分区策略实现

引言

Kafka 的分区策略决定了生产者发送的消息会被分配到哪个分区中，合理的分区策略有助于实现负载均衡、提高消息处理效率以及满足特定的业务需求。

轮询策略（默认）

• 轮询策略是 Kafka 默认的分区策略（当消息没有指定键时）。生产者会按照顺序依次将消息发送到各个分区中，确保每个分区都能均匀地接收到消息，从而实现负载均衡。简单高效，能使各个分区的消息量相对均衡，充分利用每个分区的存储和处理能力。

• import org.apache.kafka.clients.producer.*;
  import java.util.Properties;
  
  public class RoundRobinProducer {
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  
          Producer<String, String> producer = new KafkaProducer<>(props);
          for (int i = 0; i < 10; i++) {
              ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);
              producer.send(record);
          }
          producer.close();
      }
  }

  随机策略

• 随机策略会随机地将消息分配到一个分区中。这种策略在某些情况下可以实现一定程度的负载均衡，但由于是随机分配，可能会导致分区之间的消息分布不够均匀。可以通过自定义分区器来实现随机策略。

• import org.apache.kafka.clients.producer.*;
  import java.util.List;
  import java.util.Map;
  import java.util.Random;
  
  public class RandomPartitioner implements Partitioner {
      private final Random random = new Random();
  
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
          return random.nextInt(partitions.size());
      }
  
      @Override
      public void close() {}
  
      @Override
      public void configure(Map<String, ?> configs) {}
  }
  
  // 使用随机分区器的生产者示例
  public class RandomProducer {
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("partitioner.class", "RandomPartitioner");
  
          Producer<String, String> producer = new KafkaProducer<>(props);
          for (int i = 0; i < 10; i++) {
              ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);
              producer.send(record);
          }
          producer.close();
      }
  }

  按键哈希策略

• 当消息指定了键时，Kafka 会根据键的哈希值将消息分配到特定的分区中。相同键的消息会被分配到同一个分区，这有助于保证具有相同业务逻辑的消息顺序性。可以保证消息的局部有序性，例如在处理用户相关的消息时，将同一个用户的消息发送到同一个分区，方便后续的处理和分析。

• import org.apache.kafka.clients.producer.*;
  import java.util.Properties;
  
  public class KeyBasedProducer {
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  
          Producer<String, String> producer = new KafkaProducer<>(props);
          for (int i = 0; i < 10; i++) {
              ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "user-" + (i % 2), "message-" + i);
              producer.send(record);
          }
          producer.close();
      }
  }

  自定义分区策略(实现接口)

• 当上述默认策略无法满足业务需求时，可以自定义分区策略。通过实现org.apache.kafka.clients.producer.Partitioner接口，重写partition方法来实现自定义的分区逻辑。例如，根据消息的某些特定字段（如时间、地理位置等）来进行分区，以满足特定的业务需求。

• import org.apache.kafka.clients.producer.*;
  import java.util.List;
  import java.util.Map;
  
  public class CustomPartitioner implements Partitioner {
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
          // 自定义分区逻辑，这里简单示例根据消息值的长度分区
          String message = (String) value;
          return message.length() % partitions.size();
      }
  
      @Override
      public void close() {}
  
      @Override
      public void configure(Map<String, ?> configs) {}
  }
  
  // 使用自定义分区器的生产者示例
  public class CustomProducer {
      public static void main(String[] args) {
          Properties props = new Properties();
          props.put("bootstrap.servers", "localhost:9092");
          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          props.put("partitioner.class", "CustomPartitioner");
  
          Producer<String, String> producer = new KafkaProducer<>(props);
          for (int i = 0; i < 10; i++) {
              ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "message-" + i);
              producer.send(record);
          }
          producer.close();
      }
  }