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取模的方式
  Listpartitions = 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取模的方式
  Listpartitions = 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 技术支持

你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧