kafka Partition使用详解

写在前面

• kafka将主题（Topic）划分为多个不同的片段，每个片段称为一个分区

• 同一个分区的消息是有序的

• 分区可以分布在不同的存储节点上，从而提升吞吐量

• 可以通过增加分区的数量来扩展系统的处理能力

文章脉络

分区数量配置

发送消息指定分区

根据Key 确定分区

自定义分区器

Partition 分区 简介

• 默认每个Topic只有1个分区（num.partitions）

• 生产者发送消息时可以直接指定分区序号

• 消息未指定分区序号，则默认分区器确定分区序号

  • 消息Key 未指定，则轮询发送到各个分区

  • 消息Key 指定了，则根据Key 哈希确定分区序号

配置分区数量

修改 server.properties 中 num.partitions,重启生效。

发送消息时指定分区

直接在消息中指定，代码如下

   ProducerRecord producerRecord = new ProducerRecord(topic,partition,null, content);

这种方式笔者不推荐使用，理由如下

• 指定分区很容易造成不同分区数据不均匀

• 如果集群分区数量发生变化，代码可能需要跟着修改，耦合严重

消息Key 哈希确定

默认的分区器DefaultPartitioner  当Key 不为空时，根据Key 的MurmurHash2 算法计算哈希确定

如果我们希望消息有序，我们可以给消息设置相同的Key。

比如同一个商品的订单消息，我希望它是有序的被处理，那么我们可以指定订单消息的Key 为 商品id ,这样同一个商品的订单消息会发到相同的分区

 ProducerRecord producerRecord = new ProducerRecord(topic,null,"商品id", content);

随机分区

如果我们业务不关心消息存储的具体分区，则发送消息时不指定Key，也不指定分区

 ProducerRecord producerRecord = new ProducerRecord(topic, content);

自定义分区器

kafka 提供了自定义分区的功能，只需要实现Partitioner接口,以下是一个简单的分区器。

• 如果key 不为空，根据key的HashCode 确定分区

• 如果key 空，则根据Value（消息对象）的HashCode 确定分区

通过配置 partitioner.class 属性使其生效

package com.codetonight;

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

import java.util.List;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        //获取分区总个数
        int numPartitions = partitions.size();

        int partition = key != null ? key.hashCode() % numPartitions : value.hashCode() % numPartitions;

        return partition;
    }

    @Override
    public void close() {

    }

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

    }
}

配置自定义分区器

    private Map<String, Object> produceConfigs() {
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "ip:9092");
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        configMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CustomPartitioner.class);
        return configMap;
    }

    @Bean
    public ProducerFactory producerFactory() {
        Map<Class<?>, Serializer> delegates = new HashMap<>();
        delegates.put(byte[].class, new ByteArraySerializer());
        delegates.put(Bytes.class, new BytesSerializer());
        delegates.put(String.class, new StringSerializer());
        delegates.put(User.class, new UserSerializer());
        return new DefaultKafkaProducerFactory<>(produceConfigs(),
                new StringSerializer(), new DelegatingByTypeSerializer(delegates));
    }