kafka生产者发送消息分区策略

前言

分区是kafka中一个很重要的部分，合理使用分区，可以提升kafka的整体性能

kafka分区好处

1、便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果

2、提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据

3、在某些情况下，可以实现顺序消费

生产者发送消息的分区策略

我们在代码中找到ProducerRecord 这个类，可以看到有很多构造方法，即我们在调用API进行消息发送时可以选择的各种send消息的方式

下面结合这个类的几个构造方法，对生产者发送消息分区策略做简单的说明

1、默认情况下，kafka生产端采用的是DefaultPartitioner

既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）

例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）。

kafkaProducer.send(new ProducerRecord<>("zcy222","congge " + i)).get();

2、指定了分区的情况下，按照指定分区发送

比如类似下面这种，指定将数据发送到 0 号分区

运行代码，通过控制台可以看到，数据都发往了 0 号分区

3、没有指明partition值但有key

没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。

来看具体的代码，我们来做3种测试，测试1，将key的值设置为 "a"

这种情况下，可以看到，消息是发送到0号分区了

再将key设置为 b，c，f等，再次验证下，就可以验证预期的效果

4、自定义分区

kafka为开发人员提供了自定义分区的方式实现某些场景下，对于分区的选择，比如，在将消息发送到topic之前，需要对数据进行清洗、过滤之后，才能发送到指定的分区，这时候选择自定义分区就是一个很好的选择，因为在自定义分区的逻辑中可以灵活的做各种逻辑上的控制

只需要2步就可以完成自定义分区的设置，第一步，自定义一个类，实现Partitioner接口，重写里面的partition方法逻辑

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class MyPartioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String message = value.toString();
        int partition ;
        if(message.contains("congge")){
            partition = 0;
        }else {
            partition = 1;
        }
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map map) {

    }
}

在本例中，过滤出消息里面包含 "congge"的消息，将发送到0号分区，其他的发往1号分区

第二步，在生产者发送的代码中，将消息发送的属性中将自定义的分区类设置进去即可

 //关联自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.congge.kafka01.selfs.MyPartioner");

完整代码

import org.apache.kafka.clients.producer.*;

import java.util.Properties;


public class ProducerMyPartion {

    public static void main(String[] args) throws Exception {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //关联自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.congge.kafka01.selfs.MyPartioner");

        // 3. 创建 kafka 生产者对象
        KafkaProducer kafkaProducer = new KafkaProducer(properties);
        System.out.println("开始发送数据");
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("zcy222","congge" + i), new Callback() {
                @Override
                public void onCompletion(Recordmetadata metadata, Exception e) {
                    if(e == null){
                        System.out.println("发送成功");
                        System.out.println("主题:" + metadata.topic());
                        System.out.println("分区:" + metadata.partition());
                    }
                }
            });
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }

}

运行上面的代码，观察下控制台和消费者窗口数据，由于消息内容中均含有congge，所以消息都成功投递到0号分区了

实际开发过程中，可以结合自身的实际需求，进行选择使用

kafka生产者发送消息分区策略

大数据系统相关栏目本月热门文章