Kafka分区策略及自定义

柔情只为你懂 2022-06-14 07:29 185阅读 0赞

## 默认分区策略 ##

默认分区策略是：取正（bytearray生成32位hash值）%numpartitions  
这个公式的结果是得到0-(numpartitions-1)间正整数的个数大致相等，也就是说kafka的默认分区策略是无论我们给定多少个分区，我们存放的数据基本上会平均的分到各个分区上。

private int defaultPartition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            int partition = 0;
            defcount.incrementAndGet();
            if (keyBytes == null) {
                int nextValue = counter.getAndIncrement();
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                if (availablePartitions.size() > 0) {
                    int part = toPositive(nextValue) % availablePartitions.size();
                    partition = availablePartitions.get(part).partition();
                } else {
                    partition = toPositive(nextValue) % numPartitions;
                }
            } else {
                partition = toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
            return partition;
      }
    
    private static int toPositive(int number) {
            return number & 0x7fffffff;
    }
    //生成32位的hash 值
    public static int murmur2(final byte[] data) {...}

## 自定义分区策略 ##

实际开发中会遇到不让数据均匀分布，如按照范围放到不同的分区中，这样就得使用自定义的分区策略了

*  直接指定分区值  
    在创建ProducerRecord时，指定消息的分区值。

int partition = 0;
        if(key<100){
            partition = 0;
        }else if(key<200){
            partition = 1;
        }else{
            partition = 2;
        }
        ProducerRecord<String,String> records = new ProducerRecord<String,String>(TOPIC,partition,key,value);
        kafkaProducer.send(records);

*  自定义分区类  
    **1.实现Partitioner 接口**

public class KafkaCustomPartitioner implements Partitioner { 
        public void configure(Map<String, ?> arg0) {}
        public void close() {}
        public int partition(String topic, Object arg1, byte[] keyBytes, Object arg3, byte[] arg4, Cluster arg5) {
            //判断topic中是否包含acpdr
            int partition = 0;
            int key = Integer.parseInt(new String(keyBytes));
            if(key<100){
                partition = 0;
            }else if(key<200){
                partition = 1;
            }else{
                partition = 2;
            }
            return partition;
        }
    }

**2.添加配置**  
partitioner.class值为自定义分区类的完整包名，这样生产者就会选择自定义的分区策略。

props.put("partitioner.class", "xx.xx.KafkaCustomPartitioner");

说明：1.客户端测试环境中，自定义分区类跟生产者类在一个项目中，不需要其他操作；2.想要自定义的分区放到kafka的服务器端环境时，需要将自定义的分区类生成jar包放到kafka环境的lib下，同样配置文件中指定完整包名。