kafka中生产者的数据分发策略

69 阅读 0 评论 46 点赞

我是靠谱客的博主发嗲小熊猫，最近开发中收集的这篇文章主要介绍kafka中生产者的数据分发策略，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Kafka所支持的分区策略:
1- 粘性分区策略(2.4版本下, 支持轮询策略) Java客户端支持, 但是Python客户端不支持
2- hash取模的策略
3- 指定分区策略
4- 随机分区策略: Python客户端是支持的 Java不支持
5- 自定义分区策略

1- 指定分区策略

public ProducerRecord(String topic, Integer partition, K key, V value) {
this(topic, partition, null, key, value, null);
}

在生产端, 构建数据承载对象的时候, 采用此构造, 即可采用指定分区的策略
分区的编号从0开始的

注意: 指定分区与defaultPartitioner类没有任何的关系

2- Hash 取模策略

2.1 创建数据承载对象的时候, 必须使用仅传递 k v的构造方法, 即可使用Hash方式
public ProducerRecord(String topic, K key, V value) {
this(topic, null, null, key, value, null);
}

注意: 当执行Hash取模分区策略, 其底层是通过一个默认的分区类来完成Hash取模计算: DefaultPartitioner(默认分区类)

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (keyBytes == null) {
return stickyPartitionCache.partition(topic, cluster);
}
// 当有key的时候, 采用Hash取模的方式
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}

说明: 在使用此种分发策略的时候, Key值一定是可变的, 千万不要固定不变

3- 粘性分区策略

3.1 创建数据承载对象的时候,只需要传递value即可, 此时采用的粘性的分区策略
public ProducerRecord(String topic, V value) {
this(topic, null, null, null, value, null);
}

注意: 当执行Hash取模分区策略, 其底层是通过一个默认的分区类来完成Hash取模计算: DefaultPartitioner(默认分区类)

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (keyBytes == null) {
// 当key为null的时候, 采用的分发的方式为stickyPartition(粘性分区)
return stickyPartitionCache.partition(topic, cluster);
}
// 当有key的时候, 采用Hash取模的方式
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
// hash the keyBytes to choose a partition
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}

4- 自定义分区策略

如何自定义分区呢? 抄抄DefaultPartitioner

1- 创建一个类, 实现 Partitioner 接口
2- 重写接口中方法: partition() 和 close方法, 主要是重写partition()
   partition方法的参数列表说明:
       String topic: 指定要写入到那个topic上
       Object key : 表示 key值
       byte[] keyBytes: 表示 key的字节数组
       Object value: 表示value的值
       byte[] valueBytes: 表示value的字节数组
       Cluster cluster : 集群对象可以帮助获取某个topic有多少个分片

3) 将自定义的分区类, 配置到生产者的配置信息中:
   key: partitioner.class
   value值:
       默认值:org.apache.kafka.clients.producer.internals.DefaultPartitioner
       修改为我们自己写的类即可

   将其放置到生产者的properties中

# 博学谷IT 技术支持