【Kafka】原理分析：Producer分发规则

港控/mmm° 2022-12-21 04:54 272阅读 0赞

消息是kafka中最基本的数据单元，在kafka中一条消息由以下三部分构成：

*  topic：消息属于哪个消息集合
 *  value：消息具体内容
 *  key：与分区算法一起决定当前消息发送到那个partition

消息在 kafka-client 的具体实现就是 ProducerRecord 类，下面是它的构造函数：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

## 1.默认分发规则 ##

默认情况下，kafka采用的是hash取模的分区算法。

*  key不为null：就直接根据分区数取模
 *  key为null：则会随机分配一个分区。这个随机是在这个参数”metadata.max.age.ms”的时间范围内随机选择一个。对于这个时间段内，如果key为 null，则只会发送到唯一的分区。这个值默认情况下是10分钟更新一次。

> 关于Metadata，简单理解就是Topic/Partition和broker的映射关系，每一个topic的每一个partition，需要知道对应的broker列表是什么，leader是谁、follower是谁。这些信息都是存储在Metadata这个类里面。

## 2.自定义分发 ##

1). 自定义分区规则，实现Partitioner接口

public class MyPartition implements Partitioner {
        
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] 
                             valueBytes, Cluster cluster) {
        
            // 通过集群cluster获取当前topic的所有分区信息
            List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
            // 获取分区数
            int numOfPartition = partitionInfos.size();
            int partitionNum = 0;
            // 如果没有传入key，就随机分配
            if (key == null) {
        
                partitionNum = new Random().nextInt(numOfPartition);
            } else {
         // 否则通过key取模得到分区
                partitionNum = Math.abs(key.hashCode()) % numOfPartition;
            }
            // 返回分区值
            return partitionNum;
        }
    
        @Override
        public void close() {
        
    
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
        
    
        }
    }

2). 在Producer添加分区策略配置（全类名）

properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.xupt.yzh.testkafka.MyPartition")

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/e10b2c5fe4c64f25b67e9e22f281aa45.png