如何知晓Kafka消费组中服务实例的分区分配？求基于Kafka的负载均衡实现方案

有两种常用方式可以获取消费组内实例的分区分配信息：

1. Kafka命令行工具 #

使用kafka-consumer-groups脚本（Windows环境为.bat）执行查询：

kafka-consumer-groups.sh --bootstrap-server <kafka集群地址> --describe --group <消费组名称>

输出结果中，HOST字段对应服务实例的地址，PARTITIONS字段列出该实例分配到的所有分区，结合TOPIC可以明确每个实例负责的主题分区。

2. 客户端API查询 #

以Java客户端为例，通过AdminClient获取消费组的详细分配信息：

Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "<kafka集群地址>");

try (AdminClient adminClient = AdminClient.create(adminProps)) {
    DescribeConsumerGroupsResult result = adminClient.describeConsumerGroups(Collections.singletonList("<消费组名称>"));
    ConsumerGroupDescription groupDesc = result.all().get().values().iterator().next();
    
    for (MemberDescription member : groupDesc.members()) {
        System.out.println("实例ID: " + member.consumerId());
        System.out.println("实例地址: " + member.host());
        System.out.println("分配的分区: " + member.assignment().topicPartitions());
    }
} catch (Exception e) {
    e.printStackTrace();
}

可以利用Kafka的消费者分区分配机制解决这个问题：每个调度任务对应一条Kafka消息，通过主题分区与消费组的绑定，确保同一个任务只会被一个服务实例执行。

实现思路 #

1. 创建专用主题：创建一个主题（比如schedule-tasks），分区数设置为服务实例数量（这里是3个）。
2. 调度触发逻辑：所有服务实例的调度器触发时，不是直接执行任务，而是向该主题发送一条消息，消息的key使用任务的唯一标识（如任务ID），确保相同任务的消息会被路由到同一个分区。
3. 消费组配置：所有服务实例作为同一个消费组的消费者订阅该主题，Kafka会自动将每个分区分配给唯一的实例，从而保证每个分区的任务只会被一个实例处理。

代码示例 #

生产者（调度触发时发送任务消息） #

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class TaskSchedulerProducer {
    public static void sendTaskTrigger(String taskId) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "<kafka集群地址>");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // 任务消息：key为任务ID，value为任务指令
            ProducerRecord<String, String> record = new ProducerRecord<>("schedule-tasks", taskId, "execute-" + taskId);
            producer.send(record).get();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 调度器触发时调用该方法
    public static void main(String[] args) {
        sendTaskTrigger("daily-report-task");
    }
}

消费者（服务实例中处理任务） #

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.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.net.InetAddress;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class TaskExecutorConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<kafka集群地址>");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "schedule-task-group"); // 同一个消费组
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // 手动提交偏移量，确保任务执行完成

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("schedule-tasks"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(java.time.Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String taskCommand = record.value();
                    System.out.printf("实例[%s]执行任务: %s%n", 
                        InetAddress.getLocalHost().getHostName(), taskCommand);
                    // 执行实际任务逻辑
                    executeTask(taskCommand);
                    // 手动提交偏移量，避免重复执行
                    Map<TopicPartition, OffsetAndMetadata> offsetMap = Collections.singletonMap(
                        new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1)
                    );
                    consumer.commitSync(offsetMap);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void executeTask(String taskCommand) {
        // 替换为实际任务执行逻辑
        if ("execute-daily-report-task".equals(taskCommand)) {
            // 生成日报、发送通知等操作
            System.out.println("完成日报任务执行");
        }
    }
}

关键说明 #

• 由于同一个消费组的消费者会独占分配到的分区，因此每个分区的任务消息只会被一个实例处理。
• 任务消息的key保证了相同任务的消息始终进入同一个分区，确保任务不会被重复执行。
• 当实例数量变化时，Kafka会自动重新平衡分区分配，无需额外配置即可实现负载均衡。

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