如何使用Kafka Consumer从多个Topic消费记录？

Hey there! Since you've already nailed single-topic Kafka consumption, expanding to multiple topics is actually really simple. Let me break down the changes you need to make, plus a full working example:

Key Change: Update the subscribe() Argument #

Your current code uses Collections.singletonList(this.topic) to target one topic. All you need to do is replace that with a collection containing all the topics you want to subscribe to—like a list with multiple topic names.

Full Working Code Example #

try {
    // Define all the topics you want to consume from
    List<String> topicsToConsume = Arrays.asList("user-events", "order-events", "payment-events");
    consumer.subscribe(topicsToConsume);
    
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            // Optional: Use record.topic() to tell which topic the message came from
            System.out.printf("Got message from Topic: %s | Value: %s%n", 
                              record.topic(), record.value());
        }
    }
} catch (WakeupException e) {
    // Handle graceful shutdown trigger
    System.out.println("Consumer is being shut down gracefully...");
} finally {
    // Always close the consumer to clean up resources
    consumer.close();
}

Quick Notes to Keep in Mind #

• Identify Source Topics: If you need to handle messages differently based on their origin, just call record.topic() on each ConsumerRecord to get the topic name.
• Dynamic Subscriptions: If you ever need to add/remove topics while the consumer is running, just re-call consumer.subscribe() with the updated list of topics—Kafka will handle rebalancing partitions automatically.
• Group ID Considerations: Make sure your group.id consumer config is set correctly; consumers in the same group will split the workload across all subscribed topics' partitions.

内容的提问来源于stack exchange，提问作者Stella