J'ai un seul consommateur Kafka qui consomme à partir de plusieurs sujets kafka. Je voudrais pouvoir utiliser une écriture par lots dans ma destination via 1 appel d'entrée / sortie pour 100 messages, mais pour pouvoir traiter par lots, tous les messages doivent provenir du même sujet.
Si j'ai plusieurs sujets (disons 5), et lorsque consumer.poll ou consumer.consume se produit, et que je reçois par exemple 100 messages à chaque sondage, y a-t-il un moyen de m'assurer qu'ils proviennent tous du même sujet, afin que ces messages puissent être écrits par lots vers la même destination? De sorte que le prochain appel .poll obtienne le sujet suivant?
3 Réponses :
Il n'est pas possible d'effectuer un sondage par sujet - vous êtes abonné à une liste de sujets et chaque sujet peut avoir plusieurs partitions. Un sondage donné récupère un objet ConsumerRecords , qui est un conteneur de ConsumerRecord . Un ConsumerRecord représente une paire KV qui appartient à l'une des partitions de l'un des sujets auxquels vous vous êtes abonné.
Kafka essaie d'attribuer une TopicPartition aux consommateurs formant un seul groupe basé sur un cédant. Si vous n'avez qu'un seul consommateur, il revendiquera toutes les partitions de tous les sujets. Ensuite, rien ne vous empêche de vous regrouper dans le code de votre application
par exemple
private void consume() {
List<String> topics = List.of("topic1", "topic2", "topic3", "topic4", "topic5");
kafkaConsumer.subscribe(topics);
while (true) {
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);
topics.forEach(s -> {
List<ConsumerRecord<String, String>> recordsPerTopicPartition = new ArrayList<>();
consumerRecords.records(s).forEach(recordsPerTopicPartition::add);
doWhatever(recordsPerTopicPartition);
});
}
}
private void doWhatever(List<ConsumerRecord<String, String>> consumerRecords) {
//process
}
Une autre façon de gérer les sujets est la suivante: chaque ProducerRecord a une méthode topic () qui renvoie le nom du sujet pour cet enregistrement. Vous pouvez ensuite regrouper par sujet et faire ce que vous voulez avec des paires de sujets et une collection d'enregistrements pour ce sujet.
Mais je suggère fortement d'utiliser des KafkaConsumer séparés pour chaque sujet distinct si vous souhaitez gérer les sujets indépendamment.
Une solution pour s'abonner à plusieurs sujets tout en interrogeant les messages sujet par sujet consiste à utiliser les méthodes de pause / reprise.
Voici un exemple:
List<String> subscription = List.of("topic-a", "topic-b");
consumer.subscribe(suubscription);
final Map<String, List<TopicPartition>> partitionsPerTopic =
consumer.assignment()
.stream()
.collect(Collectors.groupingBy(TopicPartition::topic, Collectors.toList()));
int next = 0;
consumer.pause(consumer.assignment());
// Starting consumption
while (!closed.get()) {
// Resuming consumption for next topic
final String topic = subscription.get(next);
consumer.resume(partitionsPerTopic.get(topic));
consumer.poll(Duration.ofMillis(500)).forEach( records -> {
...
});
// Pausing consumption for current topic
consumer.pause(partitionsPerTopic.get(topic));
next = (next + 1) % subscription.size();
}
Cependant, cette solution pourrait ne pas être efficace car le consommateur peut toujours avoir des messages tamponnés pour les partitions récupérables précédentes. Ces messages seront vidés et récupérés à la deuxième itération.
En d'autres termes, cette solution peut augmenter les allers-retours du réseau entre le consommateur et les courtiers.
Qu'est-ce que vous essayez de réaliser avec cela?
essayer d'être paresseux sans avoir à engendrer un consommateur pour chaque sujet (nous avons des centaines de sujets mais certains d'entre eux ne reçoivent pas de messages pendant des jours, tandis que d'autres en reçoivent quelques par seconde), mais en même temps être performant parce que le message la base de données de destination répond mieux aux écritures par lots en fonction du sujet