kafka生产者自定义partition分区规则

目录

场景说明

自定义方法

使用方法 

场景说明

在使用Kafka的时候，我们经常需要在生产者自定义一下获取partition分区的规则。下面是简单的自定义规则。

---

自定义方法

在进行自定义的时候，我们可以先看看默认的分区规则

这个类是DefaultPartitioner，实现了Partitioner接口。

这里我们需要定义修改的是partition方法

        if (keyBytes == null) {
            return stickyPartitionCache.partition(topic, cluster);
        } 
        List partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;

默认的这个方法很简单我们只需模仿默认的实现即可，我这里设置的规则是必须传key

当key为xxx的时候，进入到0分区，否则按照默认的策略。

public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if (keyBytes == null) {
            throw new IllegalArgumentException("keyBytes不能为空");
        }
        if ("xxx".equals(key)) {
            return 0;
        }
        List partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        // hash the keyBytes to choose a partition
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map configs) {

    }
}

只需要重写partition方法即可。

在使用的时候设置partitioner.class属性并指定类，就能够使用自定义的分区规则了。

下面是一个test用例

使用方法 

properties.put("partitioner.class", "cn.mystylefree.kafkademo.config.CustomPartitioner");

@SpringBootTest
public class KafkaProducerTest {
    private static final String TOPIC_NAME = "xdclass-sp-topic";
    private static final String TOPIC_NAME1 = "xdclass-sp-topic-v1";

    public static Properties getProperties() {
        Properties props = new Properties();

        props.put("bootstrap.servers", "端口:9092");
        //props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "112.74.55.160:9092");

        // 当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别,分别是0, 1，all。
        props.put("acks", "all");
        //props.put(ProducerConfig.ACKS_CONFIG, "all");

        // 请求失败，生产者会自动重试，指定是0次，如果启用重试，则会有重复消息的可能性
        props.put("retries", 0);
        //props.put(ProducerConfig.RETRIES_CONFIG, 0);

        // 生产者缓存每个分区未发送的消息,缓存的大小是通过 batch.size 配置指定的，默认值是16KB
        props.put("batch.size", 16384);

        
        props.put("linger.ms", 10);

        
        props.put("buffer.memory", 33554432);

        
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }

     
    @Test
    public void testSendWithCustomPartition() {
        Properties properties = getProperties();
        properties.put("partitioner.class", "cn.mystylefree.kafkademo.config.CustomPartitioner");
        Producer producer = new KafkaProducer(properties);
        for (int i = 0; i < 10; i++) {
            Future send = producer.send(new ProducerRecord<>(TOPIC_NAME1, "xxx", "xxx" + i), new Callback() {
                @Override
                public void onCompletion(Recordmetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.err.println("消息发送成功！" + metadata.toString());
                    } else {
                        //异常记录
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();

    }

}

好嘞解决