Kafka consumer API
kafka消费者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> {
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
String newValue ="hao--" + record.value();
return new ProducerRecord(record.topic(),newValue);
}
@Override
public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> map) {
}
}
在生产者中添加
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(MyInterceptor.class));
把分区注释
//props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,EvenPartitioner.class);
未完,待续(仅供个人参考)
更多推荐
已为社区贡献3条内容
所有评论(0)