在消息队列Kafka中,可以指定kafka的消费位置从某个游标位置开始消费,
// 创建消费者
Properties props = new Properties();
props.setProperty("bootstrap.servers", servers); // 集群服务器地址,逗号分隔
props.setProperty("max.poll.records", String.valueOf(DrawConstant.MAX_PULL_COUNT)); // 最大拉取消费消息数量
groupId = UUID.randomUUID().toString(); // 随机组ID
props.setProperty("group.id", groupId); // 设置组ID
if (fromBegining) {
// 从头开始接收
props.setProperty("auto.offset.reset", "latest"); // 设置从头开始接收
}
props.setProperty("enable.auto.commit", "false"); // 关闭自动提交offset
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // key的格式是String
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value的格式是String
// 创建kafka实例
consumer = new KafkaConsumer<>(props);
// offset 游标位置
long s = 10;
// 订阅主题
consumer.assign(Arrays.asList(new TopicPartition(topic, 0)));
consumer.seek(new TopicPartition(topic, 0), s);
其中 **auto.offset.reset **属性可以设置为 ** none earliest latest **三种 属性如下
none
如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常
earliest
在各分区下有提交的offset时:从offset处开始消费
在各分区下无提交的offset时:从头开始消费
latest
在各分区下有提交的offset时:从offset处开始消费
在各分区下无提交的offset时:从最新的数据开始消费
当消费过的数据想要进行重复消费时 需要修改 kafka consumer 的 groupId 在同一个groupId下的数据 不会被重复消费,当修改为一个新的groupId时,则认为是一个全新的消费者,可以对一个topic下的数据进行重复消费
版权归原作者 小叶同学的猫 所有, 如有侵权,请联系我们删除。