Kafka：同一消费者组中的多个实例监听同一分区内的主题

Question

我有两个kafka consumer实例，配置了相同的消费者组，并监听相同主题中的分区0。问题是我发消息到题目的时候。消息由两个实例使用，这两个实例应该不会发生，因为它们在同一个组中。我使用Spring Boot配置类来配置它们。

以下是配置:

@Bean
ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

@Bean
public ConsumerFactory<Integer, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {

    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    return props;
}

以下是听众：

@KafkaListener(topicPartitions = {@TopicPartition(topic = "${kafka.topic.orders}", partitions = "0")})
public void consume(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {

    log.info("message received at " + orderTopic + "at partition 0");
    processRecord(record, acknowledgment);
}

Answer 1

Kafka不是这样工作的；当你手动分配这样的分区（@Topic分区）时，你是在明确告诉Kafka你想从那个分区接收消息——消费者将分区分配给自己。

换句话说，通过手动分配，您将负责分配分区。

您需要使用组管理，让Kafka将主题分配给实例。

使用< code >主题= " ... "Kafka会完成这个任务。如果没有足够的主题，实例就会闲置。您至少需要与实例一样多的分区，才能让所有实例都参与进来。