springboot中如何实现kafa指定offset消费

2023-08-04 13:10:04 305

这篇文章主要介绍了springboot中如何实现kafa指定offset消费,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下

kafka消费过程难免会遇到需要重新消费的场景，例如我们消费到kafka数据之后需要进行存库操作，若某一时刻数据库down了，导致kafka消费的数据无法入库，为了弥补数据库down期间的数据损失，有一种做法我们可以指定kafka消费者的offset到之前某一时间的数值，然后重新进行消费。

首先创建kafka消费服务

@Service
@Slf4j
//实现CommandLineRunner接口，在springboot启动时自动运行其run方法。
publicclassTspLogbookAnalysisServiceimplementsCommandLineRunner{
@Override
publicvoidrun(String...args){
//dosomething
}
}

kafka消费模型建立

kafkaserver中每个主题存在多个分区（partition），每个分区自己维护一个偏移量（offset），我们的目标是实现kafkaconsumer指定offset消费。

在这里使用consumer-->partition一对一的消费模型，每个consumer各自管理自己的partition。

@Service
@Slf4j
publicclassTspLogbookAnalysisServiceimplementsCommandLineRunner{
//声明kafka分区数相等的消费线程数，一个分区对应一个消费线程
privatestaticfinalintconsumeThreadNum=9;
//特殊指定每个分区开始消费的offset
privateListpartitionOffsets=Lists.newArrayList(1111,1112,1113,1114,1115,1116,1117,1118,1119);

privateExecutorServiceexecutorService=Executors.newFixedThreadPool(consumeThreadNum);

@Override
publicvoidrun(String...args){
//循环遍历创建消费线程
IntStream.range(0,consumeThreadNum)
.forEach(partitionIndex->executorService.submit(()->startConsume(partitionIndex)));
}
}

kafkaconsumer对offset的处理

声明kafkaconsumer的配置类

privatePropertiesbuildKafkaConfig(){
PropertieskafkaConfiguration=newProperties();
kafkaConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.GROUP_ID_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"");
kafkaConfiguration.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"");
...更多配置项

returnkafkaConfiguration;
}

创建kafkaconsumer，处理offset，开始消费数据任务#

privatevoidstartConsume(intpartitionIndex){
//创建kafkaconsumer
KafkaConsumerconsumer=newKafkaConsumer<>(buildKafkaConfig());

try{
//指定该consumer对应的消费分区
TopicPartitionpartition=newTopicPartition(kafkaProperties.getKafkaTopic(),partitionIndex);
consumer.assign(Lists.newArrayList(partition));

//consumer的offset处理
if(collectionUtils.isNotEmpty(partitionOffsets)&&partitionOffsets.size()==consumeThreadNum){
LongseekOffset=partitionOffsets.get(partitionIndex);
log.info("partition:{},offsetseekfrom{}",partition,seekOffset);
consumer.seek(partition,seekOffset);
}

//开始消费数据任务
kafkaRecordConsume(consumer,partition);
}catch(Exceptione){
log.error("kafkaconsumeerror:{}",ExceptionUtils.getFullStackTrace(e));
}finally{
try{
consumer.commitSync();
}finally{
consumer.close();
}
}
}

消费数据逻辑，offset操作

privatevoidkafkaRecordConsume(KafkaConsumerconsumer,TopicPartitionpartition){
while(true){
try{
ConsumerRecordsrecords=consumer.poll(TspLogbookConstants.POLL_TIMEOUT);
//具体的处理流程
records.forEach((k)->handleKafkaInput(k.key(),k.value()));

//

springboot中如何实现kafa指定offset消费

热门推荐

随机推荐