Java实现Kafka指定(offset)位置进行消费

在消息队列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下的数据进行重复消费