Kafka Producer - 分区机制实战

news2025/1/20 14:49:10

Kafka Producer - 分区机制实战

上一篇介绍了kafka Producer 生产者发送数据的程序代码，以及对生产者分区机制的相关介绍，今天继续深入的了解下分区机制的原理、测试验证、自定义分区。

在学习之前先在本地机器搭建一个单机版的双节点集群环境，方便后面做测试，另外本机使用的软件版本信息如下：

JDK17
kafka_2.13-3.3.1

搭建集群

修改配置

# 1. 在kafka根目录 创建cluster目录
mkdir cluster
# 2. 复制配置文件 模板
cp config/server.properties cluster/server_n1.properties
cp config/server.properties cluster/server_n2.properties
cp config/server.properties cluster/server_n3.properties

# 修改 server_n1.properties 并保存
vim cluster/server_n1.properties
# 更改相关内容如下
broker.id=0
listeners=PLAINTEXT://:9092
log.dirs=/tmp/node1/kafka-logs

# 修改 server_n2.properties 并保存
vim cluster/server_n2.properties
# 更改相关内容如下
broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/tmp/node2/kafka-logs

# 修改 server_n3.properties 并保存
vim cluster/server_n3.properties
# 更改相关内容如下
broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/tmp/node3/kafka-logs

启动集群

# 1. 启动zk
./bin/zookeeper-server-start.sh config/zookeeper.properties
# 2. 启动Node1
./bin/kafka-server-start.sh cluster/server_n1.properties
# 3. 启动Node2 
./bin/kafka-server-start.sh cluster/server_n2.properties
# 3. 启动Node3
./bin/kafka-server-start.sh cluster/server_n3.properties

创建topic

# 1. 创建一个分区为3的topic
./bin/kafka-topics.sh --create --topic topic_t3 --bootstrap-server localhost:9092 --partitions 3

# 2. 创建完成后，查看主题信息
./bin/kafka-topics.sh --describe --topic topic_t3 --bootstrap-server localhost:9092

在这里插入图片描述

测试代码

public class SimpleProducer {

    public static void main(String[] args) throws Exception{
        String topicName = "topic_t3";
        Properties props = new Properties();
        //指定kafka 服务器连接地址
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        // 发送失败 重试次数
        props.put("retries", 0);
        // 消息发送延迟时间 默认为0 表示消息立即发送，单位为毫秒
        props.put("linger.ms", 0);
        // 序列化方式
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);
        for(int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(topicName, "message : " + i);
                Future<RecordMetadata> send = producer.send(record);
                RecordMetadata metadata = send.get();
                System.out.println(String.format("sent record(key=%s value=%s) 分区数: %d, 偏移量: %d, 时间戳: %d",
                        record.key(), record.value(),
                        metadata.partition(),metadata.offset(), metadata.timestamp()
                ));
        }
        System.out.println("Message sent successfully");
        producer.close();
    }
}

无分区无Key

ProducerRecord<String, String> record = new ProducerRecord<>(topicName, "message : " + i);

注意上面代码，没有指定分区，Record对象也没有指定key值, Kakfa为了内部的性能考虑，会选取其中一个节点进行发送(避免多节点发送数据造成性能损耗)，该机制被称为黏性分区

在这里插入图片描述

无分区指定Key

// ProducerRecord<String, String> record = new ProducerRecord<>(topicName, "message : " + i);
// 替换成以下代码
ProducerRecord<String, String> record = new ProducerRecord<>(topicName, String.valueOf(i) ,"message : " + i);

重新执行，测试结果

在这里插入图片描述

当key不为空时，Kafka默认使用key的hash值，来计算待发送的分区值，核心代码如下

org.apache.kafka.clients.producer.internals.BuiltInPartitioner#partitionForKey

/*
     * Default hashing function to choose a partition from the serialized key bytes
     */
public static int partitionForKey(final byte[] serializedKey, final int numPartitions) {
  return Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions;
}

Round-Robin 机制

// 指定轮训分区算法，将数据均匀的打散在不同的节点
props.put("partitioner.class","org.apache.kafka.clients.producer.RoundRobinPartitioner");

在这里插入图片描述

测试结果如上，整体的效果图如下

在这里插入图片描述

自定义分区

程序代码

开发者可以选择实现 org.apache.kafka.clients.producer.Partitioner 接口来自定义分区机制，满足特殊的业务场景需求，接下来利用随机算法实现一个自定义分区的功能。该实现么有任何实际的作用，仅仅只是作为学习使用

public class RandomPartitioner implements Partitioner {
    private Random random = new Random();

    public void configure(Map<String, ?> configs) {}

    @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();
        return random.nextInt(numPartitions);
    }

    public void close() {}
}

测试验证

// 指定自定义的分区策略实现类
props.put("partitioner.class","org.kafka.example.RandomPartitioner");

在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/105975.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！