Kafka - 指定offset进行消费

╰+哭是因爲堅強的太久メ 2022-05-25 23:38 701阅读 0赞

# Kafka - 指定offset进行消费 #

--------------------

在网上搜索之后发现了，从消息队列最开始的位置对数据进行消费，主要代码如下：

String topicName = "A25";
    //用于分配topic和partition
    consumer.assign(Arrays.asList(new TopicPartition(topicName, 0)));
    //不改变当前offset，指定从这个topic和partition的开始位置获取。
    consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName, 0)));

轮询20次，查看数据的消费结果:

for (int i = 0; i < 20; i++) {
        ConsumerRecords<String, String> records = consumer.poll(100);
    
        logger.info("records length = {}", records.count());
    
        for (ConsumerRecord record : records) {
            logger.info("topic = {}, partition = {}, offset = {}, key = {}, value = {}\n",
                    record.topic(), record.partition(), record.offset(),
                    record.key(), record.value());
        }
    }

![kafka从头消费.png-51.1kB][kafka_.png-51.1kB]

从控制台的输出情况可以看到，kafka的消费者已经从头对数据进行了消费。同时也会发现，在拉取数据的时候会有很多次一条数据也没有拉下来，所以如果不是使用的死循环，本例中的20次轮询也可能会一条数据也消费不到。

在阅读他人博文的时候发现了有`seekToBeginning`方法时候，使用maven下载了对应jar包的源码，进行源码的查看**（不得不说，还是需要善用工具）**：

![maven下载源码.png-88.9kB][maven_.png-88.9kB]

发现还有两个其他的方法：

/**
     * Overrides the fetch offsets that the consumer will use on the next {@link #poll(long) poll(timeout)}. If this API
     * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that
     * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets
     *
     * @throws IllegalArgumentException if the provided TopicPartition is not assigned to this consumer
     *                                  or if provided offset is negative
     */
    @Override
    public void seek(TopicPartition partition, long offset)

对指定的offset处进行消费。

/**
     * Seek to the last offset for each of the given partitions. This function evaluates lazily, seeking to the
     * final offset in all partitions only when {@link #poll(long)} or {@link #position(TopicPartition)} are called.
     * If no partitions are provided, seek to the final offset for all of the currently assigned partitions.
     * <p>
     * If {@code isolation.level=read_committed}, the end offset will be the Last Stable Offset, i.e., the offset
     * of the first message with an open transaction.
     *
     * @throws IllegalArgumentException if {@code partitions} is {@code null} or the provided TopicPartition is not assigned to this consumer
     */
    public void seekToEnd(Collection<TopicPartition> partitions)

直接到消息的末尾。

但是从方法的参数中我们也可以发现：

1.  `seek(TopicPartition partition, long offset)`方法只能一个主题
2.  `seekToBeginning`和`seekToEnd`可以接受主题和分区的列表

[kafka_.png-51.1kB]: /images/20220526/d489a978b04140f9995e7251c839d300.png
[maven_.png-88.9kB]: /images/20220526/1c601da2d54b40b5bb6041cd5f61e13a.png