Kafka consumer API

创建项目

package edu.hao.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;
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.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class KafkaConsumerDemo {public static void main(String[] args) throws Exception{Properties props = new Properties();props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.74.139:9092");props.put(ConsumerConfig.GROUP_ID_CONFIG,"song");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList("song"));//订阅主题while (true){ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));for (ConsumerRecord<String,String> record :records){System.out.println(record.value());}}//consumer.close();死循环}
}

 接上一章的一起运行

一个生产数据，一个消费数据 

消费完的数据保存到了

 可以看一下这个数据

在另一台服务器中查看

kafka-console-consumer.sh --topic _consumer_offsets --bootstrap-server 192.168.74.139:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties
现在把自动消费停止后

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

不保存偏移，数据会不停的消费

可以手动保存

在while循环里for循环外

            //consumer.commitAsync();//分为同步保存和异步保存，自动最高，其次异步

重置offset

切换分组添加

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//重置offset

就可以全部消费数据

全新组设置，偏移量会保存

拦截器

创建

package edu.hao.interceptor;import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.Map;public class MyInterceptor implements ProducerInterceptor<String,String> {@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {String newValue ="hao--" + record.value();return new ProducerRecord(record.topic(),newValue);}@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> map) {}
}

 在生产者中添加

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(MyInterceptor.class));

把分区注释

//props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,EvenPartitioner.class);

未完，待续（仅供个人参考）