Kafka学习（8）- kafka使用Java开发-4-自定义分区

墨蓝 2022-03-27 04:17 232阅读 0赞

**目录：**

1.  如何在某个主题下的某个分区拿数据（假设my-topic有3个分区）
2.  为什么要自定义分区
3.  自定义分区操作流程
    
    1.  创建自定义类，实现org.apache.kafka.clients.producer.Partitioner接口
    2.  重写public int partition方法
    3.  配置项中加入partitioner.class属性

### 一.在某个主题的特定分区拿数据 ###

1.  生产者生产的数据，在分配的过程有什么原则？  
    生产者代码：
    
        package kafkaTest;
        
        import java.util.Properties;
        
        import org.apache.kafka.clients.producer.KafkaProducer;
        import org.apache.kafka.clients.producer.Producer;
        import org.apache.kafka.clients.producer.ProducerRecord;
        
        import kafkaTest.util.PropertiesUtil;
        
        public class ProducerSend {
        	public static void main(String args[]) {
        		
        		//1.属性配置：端口、缓冲内存、最大连接数、key序列化、value序列化等等
        		 /*
        		 Properties props=new Properties();
        		 props.put("bootstrap.servers", "localhost:9092");
        		 props.put("acks", "all");
        		 props.put("retries", 0); 
        		 props.put("batch.size", 16384); 
        		 props.put("linger.ms", 1);
        		 props.put("buffer.memory", 33554432); 
        		 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
        		 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");*/
        		
        		
        		//2.创建生产者对象，并建立连接,通过我们自己创建的 PropertiesUtil 工具类获得配置文件（根据传入的值，获得相对应的properties文件）
        		Producer<String, String> producer = new KafkaProducer<>(PropertiesUtil.getProperties("producer"));
        		
        		//3.在my-topic主题下，发送消息
        		for(int i = 0; i < 100; i++) {
        			//1.可根据主题和内容发送，当没有key的时候，生产者生产的数据会放在该主题的所有分区均衡分布。
        			//2.可根据主题、key和内容发送，当有key的时候，生产者生产的数据会放在同一个分区。
        			//3.可根据主题、分区、key和内容发送
        			//4.可根据主题、分区、时间戳、key和内容发送
        			producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
        			System.out.println("消息"+i);
        		}
        		
        		//4.关闭
        		producer.close();
        		
        	}
        }
    
    ![20190112000312512.png][]
2.  消费者拿到的数据是什么形式？![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70][]
    
    1.  如果主题只有一个分区，消费者拿到的数据就是生产者生产的所有的有序数据（数据1，2，3，4.....10)
    2.  如果主题下有多个分区，生产者没有根据key发送，消费者指定了特定的分区，那么消费的数据就是生产者生产的部分数据（可能是数据1,4,7,10或者2,5,8，或者3,6,9）
    3.  如果主题下有多个分区，生产者没有根据key发送，消费者没有指定特定的分区，那么消费的数据就是生产者生产的所有数据，无序数据，随机消费哪个区（数据1,3,2......或者1,2,3......或者3,2,1......或者2,3,1......等等)
    4.  如果主题下有多个分区，生产者根据key发送，消费者没有指定特定的分区，那么消费的数据就是生产者生产的所有数据（数据1，2，3，4.....10)
    5.  如果主题下有多个分区，生产者根据key发送，消费者指定特定的分区，那么消费的数据可能是生产者生产的所有数据，或者是没有（因为数据可能没有放在该分区）（数据1，2，3，4.....10或者无)  
        消费者代码：
        
            package kafkaTest;
            
            import java.util.Arrays;
            import java.util.Date;
            import java.util.Properties;
            
            import org.apache.kafka.clients.consumer.ConsumerRecord;
            import org.apache.kafka.clients.consumer.ConsumerRecords;
            import org.apache.kafka.clients.consumer.KafkaConsumer;
            import org.apache.kafka.common.TopicPartition;
            
            import kafkaTest.util.PropertiesUtil;
            
            public class ConsumerReceive {
            	public static void main(String args[]) {
            		
            		//1.参数配置:不是每一非得配置
            		/*Properties props = new Properties();
            	    props.put("bootstrap.servers", "localhost:9092");
            	    props.put("auto.commit.interval.ms", "1000");
            	    props.put("group.id", "test");//因为每一个消费者必须属于某一个消费者组，所以必须还设置group.id
            	    props.put("enable.auto.commit", "true");
            	    props.put("session.timeout.ms", "30000");
            	    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            	    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");*/
            	    
            	    //2.创建消费者对象，并建立连接
            	    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(PropertiesUtil.getProperties("consumer"));
            	    
            	    //3.设置从"my-topic"主题下拿取数据
            	    //consumer.subscribe(Arrays.asList("my-topic"));
            	    
            	    //3.设置从my-topic主题的0号分区拿数据
            	    TopicPartition tp01=new TopicPartition("my-topic", 0);
            	    consumer.assign(Arrays.asList(tp01));
            	    
            	    //4.消费数据
            	    while (true) {
            	    	//阻塞时间，从kafka中取出100毫秒的数据，有可能一次性去除0-n条
            	        ConsumerRecords<String, String> records = consumer.poll(100);
            	        //遍历
            	        for (ConsumerRecord<String, String> record : records)
            	        	//打印结果
            	                //System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
            	        	System.out.println("消费者消费的数据为："+record.value()+"-"+new Date());
            	        	
            	        	//手动提交
            	        	consumer.commitAsync();
            	    }
            	}
            }
        
        ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1][]

## ##

### 二.为什么要自定义分区 ###

假设有这么一个需求：需要把生产的数据按省份分配到10个分区，很明显，分区不够，所以自定义分区来解决这个问题

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2][]

## ##

### 三自定义分区操作流程 ###

1.  创建自定义类，实现org.apache.kafka.clients.producer.Partitioner接口
2.  重写public int partition方法
    
        package kafkaTest;
        
        import java.util.Map;
        
        import org.apache.kafka.clients.producer.Partitioner;
        import org.apache.kafka.common.Cluster;
        
        public class MyPartition implements Partitioner{
            //这个方法就决定了消息往哪个分区里面发送
            //这个方法的返回值就是表示我们的数据要去哪个分区，如果返回值是0，表示我们的数据去0分区
            public int partition(String topic, Object key, byte[] bytes, Object value, byte[] bytes1, Cluster cluster) {
                System.out.println(topic);
                System.out.println(value.toString());
                return 0;
            }
        
            public void close() {
        
            }
        
            public void configure(Map<String, ?> map) {
        
            }
        }
3.  加入partitioner.class属性
    
    1.  方法一：没有使用配置文件话，直接在生产者代码中添加
        
            props.put("partitioner.class", "KafkaTest.MyPartitioner");
        
        ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3][]
    2.  方法二：使用配置文件，添加
        
            partitioner.class=KafkaTest.MyPartitioner
        
        ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4][]

[20190112000312512.png]: /images/20220327/f4064dc67bc24cfba2b32bf24ed0f3b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70]: /images/20220327/11df056e36b3412d97b58a852e6fd0e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 1]: /images/20220327/b6c3077d929f40bfa9a660a93334ea92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 2]: /images/20220327/6e5cc5f946b5421aad91a5fa3f9949e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 3]: /images/20220327/61a9a7d7cf124ace95ec653c01c3805d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lpZ3VhbmdfODIw_size_16_color_FFFFFF_t_70 4]: /images/20220327/85c779d5e6b54c4e86786aa70b78eac8.png