Kafka自定义分区

Posted cowboys

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka自定义分区相关的知识,希望对你有一定的参考价值。

Kafka中,topic是逻辑上的概念,而partition是物理上的概念。不用担心,这些对用户来说是透明的。生产者(producer)只关心自己将消息发布到哪个topic,而消费者(consumer)只关心自己订阅了哪个topic上的消息,至少topic上的消息分布在哪些partition节点上,它本身并不关心。

如果没有分区的概念,那么topic的消息集合将集中于某一台服务器上,单节点的存储性能马上将成为瓶颈,当访问该topic存取数据时,吞吐也将成为瓶颈。

介于此,kafka的设计方案是,生产者在生产数据的时候,可以为每条消息人为的指定key,这样消息被发送到broker时,会根据分区规则,选择消息将被存储到哪一个分区中。

如果分区规则设置合理,那么所有的消息将会被均匀/线性的分布到不同的分区中,这样就实现了负载均衡和水平扩展。另外,在消费者端,同一个消费组可以多线程并发的从多个分区中 同时消费数据。 上述分区规则,实际上是实现了kafka.producer.Partitioner接口的一个类,这个实现类可以根据自己的业务规则进行自定义制定,如根据hash算法指定分区的分布规则。 如以下这个类,我们先获取key的hashcode值,再跟分区数量(配置文件中为numPartitions)做模运算,结果值作为分区存储位置,这样可以实现数据均匀线性的分布。

 


①自定义TxcPartitioner 类
public class TxcPartitioner implements Partitioner{

    @Override
    public void configure(Map<String, ?> arg0) {
        
    }

    @Override
    public void close() {
        
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int size = partitions.size();
    //如果消息的 key 为 null,默认分配到指定分区
if(keyBytes == null) { return 0; }
     //
如果 key 不为 null,并且使用了默认的分区器,kafka 会使用自己的 hash 算法对 key 取 hash 值,
//使用 hash 值与 partition 数量取模,从而确定发送到哪个分区。
//注意:此时 key 相同的消息会发送到相同的分区(只要 partition 的数量不变化)
    return Utils.toPositive(Utils.murmur2(keyBytes)) % size; 
}

②发送消息的方法如下

public void send(String topic,String key,RequestMessage message){  
        try {  
            if(kafkaProducer != null) {
                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic,key,JSONObject.toJSONString(message));  
                Future<RecordMetadata> future = kafkaProducer.send(record);
                RecordMetadata metadata = future.get();
                if(metadata != null) {
                    sysLog.debug("【Kafka message send success,topic = {}, partition is {} 】 " , metadata.topic(),metadata.partition());
                }else {
                    sysLog.error("【Kafka message send fail 】");
                    throw new KafkaSendException("Kafka message send fail");
                }
            }else {
                sysLog.error("【TxcProducer is not init】");
                throw new KafkaInitException("TxcProducer is not init");
            }
           
        }catch(Exception e){ 
            sysLog.error("【Kafka message send fail , exception = {}】 ",ExceptionUtil.collectExceptionStackMsg(e));
            throw new KafkaSendException("Kafka message send fail");  
        }  
    }  

③生产者配置中添加配置

//设置自定义分区
properties.put(TxcParameType.partitioner_class.getName(), TxcPartitioner.class.getName());


















以上是关于Kafka自定义分区的主要内容,如果未能解决你的问题,请参考以下文章

kafka 自定义分区器

kafka自定义分区规则

kafka2.5.0自定义分区器

Flink实战系列Flink SQL 写入 kafka 自定义分区策略

Kafka自定义分区

Flume自定义分区实现kafka有序