版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/u012745028/article/details/85264710
kafka中生成者自定义分区器，以及分区的分发策略，先来直接看看如何实现，直接附上代码案例
首先先实现 Partitioner接口，创建一个自定义分区器
package com.hj.kafka.producer;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
public class CustomerParatitioner implements Partitioner {
@Override
public void configure(Map<String, > configs) {
}
@Override
public int partition(String topic, Object key, byte[] keyBytes,
Object value, byte[] valueBytes, Cluster cluster) {
//key不能空 在源码中 key为空的会通过轮询的方式 选择分区
if(keyBytes == null || (!(key instanceof String))){
throw new RuntimeException("key is null");
}
//这里通过主题名称得到该主题所有的分区信息
List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
int numPartition = partitionInfos.size();
if(numPartition <= 1){
return 0;
}
//后面进行逻辑判断返回哪个分区, 这里比如 key为123的 选择放入最后一个分区
if(key.toString().equals("123")){
return partitionInfos.size()-1;
}
//返回源码中默认的 按照原地址散列
return (Math.abs(Utils.murmur2(keyBytes)) % (numPartition - 1));
}
@Override
public void close() {
}
}
上面代码可以看到，我们实现了一个自定的分区器，当消息的 key为字符串123的时候，我们选择放入到最后一个分区当中（我们假设创建一个主题的时候创建了五个分区）, 我们自定义的分区器key不能为空，否则抛出异常，在默认的分区器中，如果key为空的时候是通过内部的轮询的方式来选择分区的，如果key不为空的情况下通过内部的一致性hash算法来选择分区，这里的hash算法是kafka自己实现的，所以jdk升级都不会有影响，如何选择分区的参考KafkaProducer的partition方法，查看我这里的注释
private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
//这里是通过返送消息 构建ProducerRecord的时候指定了分区的情况 直接找到主题对应的分区就行
Integer partition = record.partition();
if (partition != null) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
int numPartitions = partitions.size();
// they have given us a partition, use it
if (partition < 0 || partition >= numPartitions)
throw new IllegalArgumentException("Invalid partition given with record: " + partition
+ " is not in the range [0..."
+ numPartitions
+ "].");
return partition;
}
//这里就会调用 我们自定义的分区 或者 kafka默认的分区
return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue,
cluster);
}
然后默认分区的源码可以看看DefaultPartitioner的partition方法
/**
*
* @param topic
* @param key
* @param keyBytes
* @param value
* @param valueBytes
* @param cluster
* @return
*/
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();
//key为空的情况
if (keyBytes == null) {
//这里通过得到一个原子类型的整形类型的自增值
int nextValue = counter.getAndIncrement();
//根据主题名称得到所有可用分区信息
List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
//下面根据自增值对分区数量进行取余操作， 实现了一个轮询获取分区的效果
if (availablePartitions.size() > 0) {
int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
return availablePartitions.get(part).partition();
} else {
// no partitions are available, give a non-available partition
return DefaultPartitioner.toPositive(nextValue) % numPartitions;
}
} else {
//这里通过key进行kafka的一致性hash算法 对分区数量取余得到具体分区
return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}
上面两段代码可以看到构建消息的时候如果指定了分区，直接得到分区，如果没有指定分区 key为空 那么根据轮询个的方式得到分区，如果key不为空，根据kafka内部的一致性的hash散列算法得到分区
下面看看生产者代码如何指定之定义分区器
生产者代码如下
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class ProducerTest1 {
public void execMsgSend() throws Exception{
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.66.138:9092,192.168.66.139:9092,192.168.66.140:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("buffer.memory", 33554432);
//这个地方指定了分区器，如果不指定就是默认的
props.put("partitioner.class", "com.hj.kafka.producer.CustomerParatitioner");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//生产者发送消息
String topic = "test6666";
Producer<String, String> procuder = new KafkaProducer<String,String>(props);
for (int i = 1; i <= 1000; i++) {
String value = "value_" + i;
//构建生产记录， 这里指定了key为123
ProducerRecord<String, String> msg = new ProducerRecord<String, String>(topic, "123", value);
RecordMetadata metadata = procuder.send(msg).get();
//打印分区信息
String result = "value [" + msg.value() + "] has been sent to partition " + metadata.partition();
System.out.println(result);
Thread.sleep(500);
}
System.out.println("send message over.");
procuder.close(100,TimeUnit.MILLISECONDS);
}
public static void main(String[] args) throws Exception {
ProducerTest1 test = new ProducerTest1();
test.execMsgSend();
}
}
从上面生产者代码可以看出 自定一个分区器之后 直接配置的时候指定就行， 可以去看看生产者发送的源码，可以看出如何得到分区信息，发现kafka的分区的分发策略也是挺简单的