1.创建java类MyPartitioner并实现Partitioner接口
点击灯泡选择实现方法，导入需要实现的抽象方法
2.实现方法
3.自定义分区器的使用
在自定义生产者消息发送时，属性配置上加入自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.ljr.kafka.producer.MyPartitioner");
4.运行
可以看到数据按照分区器的设定进入指定的分区
5.完整代码
package com.ljr.kafka.producer;import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;import java.util.Map;public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {//获取数据String data = value.toString();//定义分区int partition;if(data.contains("kafka")){partition = 0;}else{partition = 1;}return partition;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}
