Kafka从不同分区读取相同消息

以下是一个使用Java语言的示例代码，展示了如何从不同分区中读取相同的消息：

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

import java.util.*;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        String topicName = "your-topic-name";
        String bootstrapServers = "your-kafka-bootstrap-servers";

        // 创建消费者配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        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");

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

        // 获取主题的分区信息
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topicName);
        List<TopicPartition> partitions = new ArrayList<>();

        // 为每个分区创建TopicPartition对象
        for (PartitionInfo partitionInfo : partitionInfos) {
            partitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
        }

        // 将消费者分配给分区
        consumer.assign(partitions);

        // 从每个分区读取消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Topic: " + record.topic() +
                        ", Partition: " + record.partition() +
                        ", Offset: " + record.offset() +
                        ", Key: " + record.key() +
                        ", Value: " + record.value());
            }
        }
    }
}

在上面的代码中，我们首先创建了一个Kafka消费者，然后获取了指定主题的分区信息。然后，我们使用assign方法将消费者分配给每个分区。最后，在一个无限循环中，我们使用poll方法从每个分区中读取消息，并打印出消息的相关信息。

请确保将your-topic-name和your-kafka-bootstrap-servers替换为实际的主题名称和Kafka引导服务器地址。