【Kafka】自定义分区策略

自定义分区策略

---

思路

Command+Option+shift+N 调出查询页面，找到producer包的Partitioner接口

Partitioner下有一个DefaultPartitioner实现类

这里就有之前提到kafka数据分区策略

---

自定义分区策略

创建一个MyPartitioner类，继承并重新定义上面的Partitioner类

package cn.itcast.kafka.demo1;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

public class MyPartitioner implements Partitioner {
    /**
     * 此方法是确定分区规则
     * @param topic
     * @param key
     * @param keyBytes
     * @param value
     * @param valueBytes
     * @param cluster
     * @return 返回的int值为分区
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    	//return 3 则指定发送数据到3分区
        return 3;
    }

    @Override
    public void close() {

    }

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

    }
}

还需要在MyProducer中添加一行代码

props.put("partitioner.class","cn.itcast.kafka.demo1.MyPartitioner");

而且在MyProducer类中不需要指定分区号

producer.send(new ProducerRecord<String, String>("test" , "mykey" + i,"这是第" + i + "条message"));