kafka的生产者分区机制原理（二）

kafka分区概念

消费者给kafka发送消息的时候相同的topic可以有多个分区。且分区可以被放在不同的服务器，消费者的消息只会被发送到一个分区里，比如说某个topic有100个分区，消费者产生了100条消息，此时各个分区都有一条消息。且每个分区都会有多个副本，且以其中的一个分区为leader，其他的分区为fllower。

kafka为什么要分区?

负载均衡，实现系统的高伸缩性。为什么这么说呢？因为不同的分区可以放置在不通的机器节点上，当我们服务的吞吐量特别大的时候，可以通过增加节点来进行提高吞吐量。
实现业务逻辑上的功能：实现业务级别的消息顺序的问题。

分区策略

分区策略指的是决定生产者将消息发送到那个分区的算法。
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;
  }

2.随机策略也称 Randomness 策略。所谓随机就是我们随意地将消息放置到任意一个分区上，如下面这张图所示。

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return ThreadLocalRandom.current().nextInt(partitions.size());

按消息键保序策略也称 Key-ordering 策略,。Kafka 允许为每条消息定义消息键，简称为 Key。这个 Key 的作用非常大，它可以是一个有着明确业务含义的字符串，比如客户代码、部门编号或是业务 ID 等；也可以用来表征消息元数据。特别是在 Kafka 不支持时间戳的年代，在一些场景中，工程师们都是直接将消息创建时间封装进 Key 里面的。一旦消息被定义了 Key，那么你就可以保证同一个 Key 的所有消息都进入到相同的分区里面，由于每个分区下的消息处理都是有顺序的，故这个策略被称为按消息键保序策略，如下图所示。

List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
return Math.abs(key.hashCode()) % partitions.size();

自定义分区策略：需要显式的去配置参数partitioner.class

    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);

自定义分区的完整包名：

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