在Kafka中，消费者组是一种机制，用于将来自一个主题的消息分发给多个消费者。以下是Kafka消费者组管理的一些关键概念和命令：

关键概念

• 消费者组（Consumer Group）：一组共享同一个组ID的消费者实例，它们共同消费一个或多个主题的分区。
• 分区（Partition）：主题被分割成多个分区，每个分区存储一部分数据，消费者可以并行消费分区中的数据。
• 消费者实例（Consumer Instance）：运行中的消费者应用程序实例。
• 组ID（Group ID）：用于标识消费者组的唯一标识符。

常用命令

• 列出消费者组：

  kafka-consumer-groups.sh --bootstrap-server  --list
  

  这个命令会列出所有已注册到指定Broker的消费者组。

• 描述消费者组：

  kafka-consumer-groups.sh --bootstrap-server  --describe --group 
  

  这个命令会显示指定消费者组的详细信息，包括每个分区的消费者偏移量、LAG（日志长度）等。

• 重新平衡消费者组： Kafka会自动管理消费者组的重新平衡，但有时你可能需要手动触发。可以使用以下命令：

  kafka-consumer-groups.sh --bootstrap-server  --rebalance 
  

• 移除消费者组： 如果你想移除一个消费者组，可以使用以下命令：

  kafka-consumer-groups.sh --bootstrap-server  --remove --group 
  

配置消费者组的Java示例

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.serialization.StringDeserializer;

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

public class KafkaConsumerExample {
    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String groupId = "my-consumer-group";
        Properties consumerProps = KafkaConsumerConfig.getConsumerProperties(bootstrapServers, groupId);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Arrays.asList("my-topic"));

        try {
            while (true) {
                ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord record : records) {
                    System.out.printf("Offset = %d, Key = %s, Value = %s%n", record.offset(), record.key(), record.value());
                }
                // 手动提交偏移量
                consumer.commitSync();
            }
        } finally {
            consumer.close();
        }
    }

    public static Properties getConsumerProperties(String bootstrapServers, String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return props;
    }
}

通过这些命令和配置示例，你可以有效地管理Kafka消费者组，确保数据被均匀分配给消费者实例，并在需要时进行重新平衡或移除。