时间:2021-05-19
这篇文章主要介绍了springboot中如何实现kafa指定offset消费,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
kafka消费过程难免会遇到需要重新消费的场景,例如我们消费到kafka数据之后需要进行存库操作,若某一时刻数据库down了,导致kafka消费的数据无法入库,为了弥补数据库down期间的数据损失,有一种做法我们可以指定kafka消费者的offset到之前某一时间的数值,然后重新进行消费。
首先创建kafka消费服务
kafka消费模型建立
kafka server中每个主题存在多个分区(partition),每个分区自己维护一个偏移量(offset),我们的目标是实现kafka consumer指定offset消费。
在这里使用consumer-->partition一对一的消费模型,每个consumer各自管理自己的partition。
@Service@Slf4jpublic class TspLogbookAnalysisService implements CommandLineRunner { //声明kafka分区数相等的消费线程数,一个分区对应一个消费线程 private static final int consumeThreadNum = 9; //特殊指定每个分区开始消费的offset private List<Long> partitionOffsets = Lists.newArrayList(1111,1112,1113,1114,1115,1116,1117,1118,1119); private ExecutorService executorService = Executors.newFixedThreadPool(consumeThreadNum); @Override public void run(String... args) { //循环遍历创建消费线程 IntStream.range(0, consumeThreadNum) .forEach(partitionIndex -> executorService.submit(() -> startConsume(partitionIndex))); }}kafka consumer对offset的处理
声明kafka consumer的配置类
创建kafka consumer,处理offset,开始消费数据任务#
消费数据逻辑,offset操作
private void kafkaRecordConsume(KafkaConsumer<String, byte[]> consumer, TopicPartition partition) { while (true) { try { ConsumerRecords<String, byte[]> records = consumer.poll(TspLogbookConstants.POLL_TIMEOUT); //具体的处理流程 records.forEach((k) -> handleKafkaInput(k.key(), k.value())); //🌿很重要:日志记录当前consumer的offset,partition相关信息(之后如需重新指定offset消费就从这里的日志中获取offset,partition信息) if (records.count() > 0) { String currentOffset = String.valueOf(consumer.position(partition)); log.info("current records size is:{}, partition is: {}, offset is:{}", records.count(), consumer.assignment(), currentOffset); } //offset提交 consumer.commitAsync(); } catch (Exception e) { log.error("handlerKafkaInput error{}", ExceptionUtils.getFullStackTrace(e)); } }}以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
本文实例讲述了jQuery实现点击滚动到指定元素上的方法。分享给大家供大家参考,具体如下:jquery点击滚动到指定元素上,1.获取指定元素的offset()得
前言我们大家都知道,mysql查询使用select命令,配合limit,offset参数可以读取指定范围的记录,但是offset过大影响查询性能的原因及优化方法
前言mysql查询使用select命令,配合limit,offset参数可以读取指定范围的记录。本文将介绍mysql查询时,offset过大影响性能的原因及优化
前言在之前的SpringBoot基础教程系列中,已经通过《SpringBoot中使用@Async实现异步调用》一文介绍过如何使用@Async注解来实现异步调用了
在Flutter中如何实现点击2次Back按钮退出App,如何实现App中多个Route(路由),如何实现Back按钮只退出指定页面,此篇文章将告诉你。Will