获取Kafka消息时如何排除指定分区？

针对你提到的场景——多个Kafka主题各含2个分区，仅需消费分区0、分区1预留不用，且使用编程式配置的ConcurrentKafkaListenerContainerFactory，以下是几种可行的实现方式：

1. 编程式指定要消费的具体分区 #

直接在创建KafkaListenerEndpoint时，明确指定每个主题仅订阅分区0，从源头避免订阅到分区1。

示例代码：

// 假设你要消费的主题列表为topicsList
List<String> topicsList = Arrays.asList("topic1", "topic2", "topic3");

// 创建MethodKafkaListenerEndpoint（根据实际使用的Endpoint类型调整）
MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
endpoint.setId("custom-endpoint");
endpoint.setBean(yourConsumerBean);
endpoint.setMethod(yourConsumerMethod);

// 构造每个主题对应的分区配置，仅包含分区0
List<TopicPartition> topicPartitions = topicsList.stream()
        .map(topic -> new TopicPartition(topic, 0))
        .collect(Collectors.toList());
endpoint.setTopicPartitions(topicPartitions);

// 将endpoint注册到KafkaListenerEndpointRegistry
registry.registerListenerContainer(endpoint, containerFactory);

这种方式最直接，完全绕过自动分区分配逻辑，精准控制消费的分区。

2. 自定义分区分配策略 #

实现Kafka的PartitionAssignor接口，在分配阶段过滤掉所有主题的分区1，让消费者永远不会被分配到该分区。

步骤1：实现自定义分配策略 #

public class ReservedPartitionExcluder implements PartitionAssignor {
    @Override
    public Map<String, List<PartitionAssignment>> assign(Map<String, Integer> partitionsPerTopic, Map<String, Subscription> subscriptions) {
        // 先使用默认分配策略（比如RangeAssignor）得到初始分配结果
        PartitionAssignor defaultAssignor = new RangeAssignor();
        Map<String, List<PartitionAssignment>> defaultAssignments = defaultAssignor.assign(partitionsPerTopic, subscriptions);
        
        // 过滤掉每个分配结果中的分区1
        return defaultAssignments.entrySet().stream()
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        entry -> entry.getValue().stream()
                                .filter(assignment -> assignment.partition() != 1)
                                .collect(Collectors.toList())
                ));
    }

    @Override
    public Subscription subscription(Set<String> topics) {
        return new Subscription(topics);
    }

    @Override
    public void onAssignment(Assignment assignment) {
        // 无需额外操作
    }

    @Override
    public String name() {
        return "reserved-partition-excluder";
    }
}

步骤2：配置到消费者工厂 #

在创建ConsumerFactory时，添加分区分配策略的配置：

Map<String, Object> consumerProps = new HashMap<>();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-servers");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");
// 配置自定义分区分配策略
consumerProps.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, ReservedPartitionExcluder.class.getName());

DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);

这种方式适合需要保留自动订阅主题，但过滤特定分区的场景，即使后续新增主题，只要符合“分区1为预留”的规则，都会自动排除。

3. 容器自定义器调整订阅逻辑 #

通过ConcurrentKafkaListenerContainerFactory的容器自定义器，在容器启动前将“订阅主题”改为“订阅指定分区”，替换默认的全主题订阅。

示例代码：

ConcurrentKafkaListenerContainerFactory<String, String> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
containerFactory.setConsumerFactory(consumerFactory);

// 设置容器自定义器
containerFactory.setContainerCustomizer(container -> {
    ContainerProperties props = container.getContainerProperties();
    // 获取原本要订阅的主题
    Collection<String> topics = props.getTopics();
    if (!topics.isEmpty()) {
        // 将主题订阅替换为分区0的订阅
        List<TopicPartitionInitialOffset> partitionOffsets = topics.stream()
                .map(topic -> new TopicPartitionInitialOffset(topic, 0))
                .collect(Collectors.toList());
        props.setTopicPartitions(partitionOffsets);
        // 清空原主题订阅
        props.setTopics(null);
    }
});

这种方式适合已经通过@KafkaListener指定了主题，但需要统一修改为只消费分区0的场景，无需修改每个@KafkaListener注解。

你之前尝试的ConsumerRebalanceListener确实无法实现需求，因为Kafka不允许在onPartitionsAssigned方法中移除已分配的分区，该方法仅用于在分区分配完成后做一些初始化操作（比如重置偏移量），无法改变分配结果。

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