北京天仪建设工程质量检测所网站,Wordpress微博样式,免费注册网址域名,包装设计公司哪个好【1】分区策略
Kafka中Topic是对数据逻辑上的分类#xff0c;而Partition才是数据真正存储的物理位置。所以在生产数据时#xff0c;如果只是指定Topic的名称#xff0c;其实Kafka是不知道将数据发送到哪一个Broker节点的。我们可以在构建数据传递Topic参数的同时#xff…【1】分区策略
Kafka中Topic是对数据逻辑上的分类而Partition才是数据真正存储的物理位置。所以在生产数据时如果只是指定Topic的名称其实Kafka是不知道将数据发送到哪一个Broker节点的。我们可以在构建数据传递Topic参数的同时也可以指定数据存储的分区编号。 指定分区传递数据是没有任何问题的。Kafka会进行基本简单的校验比如是否为空是否小于0之类的。但是你的分区是否存在就无法判断了所以需要从Kafka中获取集群元数据信息此时会因为长时间获取不到元数据信息而出现超时异常。所以如果不能确定分区编号范围的情况不指定分区还是一个不错的选择。
如果不指定分区Kafka会根据集群元数据中的主题分区来通过算法来计算分区编号并设定
(1) 如果指定了分区直接使用
(2) 如果指定了自己的分区器通过分区器计算分区编号如果有效直接使用
(3) 如果指定了数据Key且使用Key选择分区的场合采用murmur2非加密散列算法类似于hash计算数据Key序列化后的值的散列值然后对主题分区数量模运算取余最后的结果就是分区编号
(4) 如果未指定数据Key或不使用Key选择分区那么Kafka会采用优化后的粘性分区策略进行分区选择
没有分区数据加载状态信息时会从分区列表中随机选择一个分区。如果存在分区数据加载状态信息时 根据分区数据队列加载状态通过随机数获取一个权重值。根据这个权重值在队列加载状态中进行二分查找法查找权重值的索引值。将这个索引值加1就是当前设定的分区。
增加数据后会根据当前粘性分区中生产的数据量进行判断是不是需要切换其他的分区。判断标准就是大于等于批次大小16K的2倍或大于一个批次大小16K且需要切换。如果满足条件下一条数据就会放置到其他分区。
【2】分区器
在某些场合中指定的数据我们是需要根据自身的业务逻辑发往指定的分区的。所以需要自己定义分区编号规则而不是采用Kafka自动设置。Kafka早期版本中提供了两个分区器不过在当前kafka版本中已经不推荐使用了。
自定义分区器
首先我们需要创建一个类然后实现Kafka提供的分区类接口Partitioner接下来重写方法。这里我们只关注partition方法即可因为此方法的返回结果就是需要的分区编号。
/*** TODO 自定义分区器实现步骤* 1. 实现Partitioner接口* 2. 重写方法* partition : 返回分区编号从0开始* close* configure*/
public class KafkaPartitionerMock implements Partitioner {/*** 分区算法 - 根据业务自行定义即可* param topic The topic name* param key The key to partition on (or null if no key)* param keyBytes The serialized key to partition on( or null if no key)* param value The value to partition on or null* param valueBytes The serialized value to partition on or null* param cluster The current cluster metadata* return 分区编号从0开始*/Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {return 0;}Overridepublic void close() {}Overridepublic void configure(MapString, ? configs) {}
}配置分区器
public class ProducerPartitionTest {public static void main(String[] args) {MapString, Object configMap new HashMap();configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);configMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());configMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 这里配置自定义分区器configMap.put( ProducerConfig.PARTITIONER_CLASS_CONFIG, KafkaPartitionerMock.class.getName());KafkaProducerString, String producer null;try {producer new KafkaProducer(configMap);for ( int i 0; i 1; i ) {ProducerRecordString, String record new ProducerRecordString, String(test, key i, value i);final FutureRecordMetadata send producer.send(record, new Callback() {public void onCompletion(RecordMetadata recordMetadata, Exception e) {if ( e ! null ) {e.printStackTrace();} else {System.out.println(数据发送成功 record.key() , record.value());}}});}} catch ( Exception e ) {e.printStackTrace();} finally {if ( producer ! null ) {producer.close();}}}
}
