Kafka单个消费者从给定偏移量的多个分区中读取主题。

以下是使用Java编写的示例代码，演示了如何使用Kafka的Consumer API从给定偏移量的多个分区中读取主题。

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // Kafka 服务器的地址
        String bootstrapServers = "localhost:9092";
        // 消费者组的ID
        String groupId = "test-group";
        // 要消费的主题
        String topic = "test-topic";
        // 要从每个分区开始消费的偏移量
        long offset = 0L;

        // 配置消费者属性
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建消费者实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 分配给消费者指定的分区
        TopicPartition partition0 = new TopicPartition(topic, 0);
        TopicPartition partition1 = new TopicPartition(topic, 1);
        consumer.assign(Arrays.asList(partition0, partition1));

        // 从指定偏移量开始消费消息
        consumer.seek(partition0, offset);
        consumer.seek(partition1, offset);

        // 持续消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            // 处理接收到的消息
            records.forEach(record -> {
                System.out.printf("Partition: %s, Offset: %d, Key: %s, Value: %s%n",
                        record.partition(), record.offset(), record.key(), record.value());
            });
        }
    }
}

以上代码创建了一个消费者实例，并通过assign方法将两个分区分配给消费者。然后使用seek方法将消费者的偏移量设置为指定的值。最后，使用poll方法持续地从分配的分区中获取消息并进行处理。