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的消费者已经从头对数据进行了消费。同时也会发现，在拉取数据的时候会有很多次一条数据也没有拉下来，所以如果不是使用的死循环，本例中的20次轮询也可能会一条数据也消费不到。

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

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

/**
 * 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可以接受主题和分区的列表