java kafka 消费者

迈不过友情╰ 2022-05-08 13:33 248阅读 0赞

**消费者消费的模式有很多种，我这里只是把消费的最简单的代码贴一下，其他的消费模式，就稍微提一下改动的**地方

public class Consumer { 
    
        private static KafkaConsumer<String,String> consumer = null;
    
        static { 
            Properties configs = initConfig();
            consumer = new KafkaConsumer<String, String>(configs);
        }
    
        private static Properties initConfig(){ 
            Properties properties = new Properties();
            properties.put("bootstrap.servers","127.0.0.1:9092");
            // 指定消费组
            properties.put("group.id","0");
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            // 设置为手动提交
            properties.setProperty("enable.auto.commit", "false");
            // earliest 分区的开头 latest 分区的结尾 必须是一个全新的消费组才可
            properties.setProperty("auto.offset.reset", "earliest");
            return properties;
        }
    
    
        public static void main(String[] args) { 
            // 根据主题消费数据
            consumer.subscribe(Arrays.asList("part"));
            //根据某个主题下的分区消费数据
      /* TopicPartition topicPartition = new TopicPartition("part",1); consumer.assign(Arrays.asList(topicPartition));*/
    
            while (true) { 
                // 隔多久拿一次数据
                ConsumerRecords<String, String> records = consumer.poll(10);
    
                for (ConsumerRecord<String, String> record : records) { 
                    System.out.println(" Consumer value:"+record.value()+";partition:"+record.partition()+";offset:"+record.offset());
    
                    // 手动提交,如是自动提交，注释即可
                   consumer.commitAsync();
                }
            }
        }
    }

注意，底下是一些关键性配置

# kafka一些配置说明 #

## 生产者消息发送到指定分区 ##

自定义分区，发送消息时，设置key,根据key判断发送到指定分区
    自定义分区参考 MyPartition.java

## 消费者消费某个主题下的分区 ##

TopicPartition topicPartition = new TopicPartition("part",1);
     consumer.assign(Arrays.asList(topicPartition));

## 指定某个分区下的offset消费 ##

注意，此情况必须要改为手动提交，要不然offset不会改变
    TopicPartition topicPartition = new TopicPartition("part",0);
    consumer.assign(Arrays.asList(topicPartition));
    consumer.seek(topicPartition,62)

## 从最后的offset消费\`\`\`\`\`\`\`\`\`\` ##

注意，此情况必须要改为手动提交，要不然offset不会改变
    如你重新订阅整个分区，以前没消费的都会消费掉
    TopicPartition topicPartition = new TopicPartition("part",0);
    consumer.assign(Arrays.asList(topicPartition));
    consumer.seekToEnd(Arrays.asList(topicPartition));

## 分区开头或结尾消费 ##

earliest 分区的开头 latest 分区的结尾 必须是一个全新的消费组才可以
    properties.put("auto.offset.reset", "earliest");

## 消费者手动提交 ##

设置 
    properties.put("enable.auto.commit", "false");
     consumer.commitAsync();