kafka配置多个消费者groupid kafka多个消费者消费同一个partition（java）

在Apache Kafka中，消费者组（consumer group）是一个重要的概念。消费者组可以让多个消费者共同消费同一个主题（topic）中的消息。每个消费者组可以拥有多个消费者，但在同一个消费者组内，每个分区（partition）只能被一个消费者消费。为了实现多个消费者消费同一个分区，我们需要让它们属于不同的消费者组。

1. 理解消费者组和分区的关系

在Kafka中，主题通常会分为多个分区。每个分区能够被配合一个消费者消费，但同一个消费者组中的多个消费者会竞争各自的分区。因此，如果多个消费者希望处理同一份数据，它们需要属于不同的消费者组。

2. 配置Kafka生产者和消费者

我们可以用Java代码来配置Kafka的生产者（Producer）和消费者（Consumer）。下面将分别给出生产者和消费者的配置示例。

2.1 生产者配置

生产者的代码比较简单，主要是设置一些基本的属性，指定主题并发送消息：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("my_topic", "key" + i, "value" + i);
            producer.send(record, (RecordMetadata metadata, Exception e) -> {
                if (e != null) {
                    e.printStackTrace();
                } else {
                    System.out.printf("Sent message with key %s to partition %d\n", record.key(), metadata.partition());
                }
            });
        }

        producer.close();
    }
}

2.2 消费者配置

消费者则需要根据其组ID进行配置。我们可以创建多个消费者，它们各自属于不同的组ID，从而实现对同一分区的消费。

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.clients.consumer.ConsumerRecord;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        String groupId = "my_group"; // 根据需要更改 groupId
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        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);
        consumer.subscribe(Collections.singletonList("my_topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Consumer Group %s consumed message from partition %d: %s\n",
                        groupId, record.partition(), record.value());
            }
        }
    }
}

3. 启动多个消费者

如果您希望模拟多个消费者消费同一条消息，只需启动多个消费者程序（可以在不同的进程中运行），并为它们指定不同的groupId：

// 启动第一个消费者
KafkaConsumerExample.main(new String[]{"my_group1"});

// 启动第二个消费者
KafkaConsumerExample.main(new String[]{"my_group2"});

如上所示，my_group1 和 my_group2 是不同的消费组。这样，这两个消费者就可以独立地消费同一主题（如 my_topic）中的消息。

4. 总结

通过使用不同的消费者组，可以轻松地实现多个消费者消费同一分区的消息。这使得Kafka在处理高并发场景时更加灵活和高效。应用此设计原则后，可以充分利用Kafka的优势，确保消息得以可靠且快速地处理。希望本文所提供的示例代码能够帮助您更好地理解Kafka消费者组的使用。