WIXLIB

d'instance client spécifique d'un groupe de consommateurs, il suffit d'affecter les groupes group.idà ce client, via la configuration du client:Properties props new Properties();props.put("group.id", "groupName");// .some more properties requirednew KafkaConsumer K, V (config);Ainsi, tous les consommateurs qui se connectent au même cluster Kafka et utilisent le mêmegroup.id forment un groupe de consommateurs. Les consommateurs peuvent quitter un groupe àtout moment et les nouveaux consommateurs peuvent rejoindre un groupe à tout moment. Dansles deux cas, un soi-disant rééquilibrage est déclenché et les partitions sont réaffectées au groupede consommateurs pour garantir que chaque partition est traitée par un seul consommateur dugroupe.Faites attention, même un seul KafkaConsumer forme un groupe de consommateurs en tant quemembre unique.Gestion des crédits à la consommation et tolérance aux fautesKafkaConsumers demande des messages à un courtier Kafka via un appel à poll() et leurprogression est suivie par des décalages . Chaque message de chaque partition de chaque sujetest associé à un décalage, c'est-à-dire son numéro de séquence logique dans la partition. UnKafkaConsumer suit son décalage actuel pour chaque partition qui lui est affectée. Faites attention,les courtiers Kafka ne sont pas au courant des compensations actuelles des consommateurs.Ainsi, sur poll() le consommateur doit envoyer ses décalages actuels au courtier, de sorte que lecourtier puisse renvoyer les messages correspondants, c.-à-d. messages avec un décalageconsécutif plus important. Par exemple, supposons que nous ayons un seul sujet de partition et unseul consommateur avec le décalage actuel 5. Sur poll() le consommateur envoie le décalage aucourtier et le courtier renvoie les messages pour les décalages 6,7,8, .Étant donné que les consommateurs suivent eux-mêmes leurs décalages, ces informationspeuvent être perdues en cas de défaillance d'un consommateur. Par conséquent, les décalagesdoivent être stockés de manière fiable, de sorte qu’au redémarrage, un consommateur puisserécupérer son ancien décalage et son nouvel article là où il l’a laissé. Dans Kafka, il existe unsupport intégré pour cela via des commits offset . Le nouveau KafkaConsumer peut valider sondécalage actuel sur Kafka et Kafka stocke ces décalages dans une rubrique spéciale appeléeconsumer offsets . Stocker les décalages dans une rubrique Kafka n'est pas seulement tolérantaux pannes, mais permet également de réaffecter des partitions à d'autres consommateurs lorsd'un rééquilibrage. Étant donné que tous les consommateurs d'un groupe de consommateurspeuvent accéder à tous les décalages validés de toutes les partitions, lors d'un rééquilibrage, unconsommateur qui reçoit une nouvelle partition lit simplement le décalage consumer offsets decette partition à partir de la rubrique consumer offsets .Comment commettre des compensationsKafkaConsumers peut valider automatiquement les décalages en arrière-plan (paramètre deconfiguration enable.auto.commit true ) quel est le paramètre par défaut. Ces validationshttps://riptutorial.com/fr/home9

automatiques sont effectuées dans poll() ( généralement appelé dans une boucle ). La fréquenceà laquelle les décalages doivent être auto.commit.interval.ms peut être configurée viaauto.commit.interval.ms . Comme les validations automatiques sont incorporées dans poll() etque poll() est appelé par le code utilisateur, ce paramètre définit une limite inférieure pourl'intervalle inter-validation.Comme alternative à la validation automatique, les décalages peuvent également être gérésmanuellement. Pour cela, la validation automatique doit être désactivée ( enable.auto.commit false ). Pour la KafkaConsumers manuelle, KafkaConsumers propose deux méthodes, à savoircommitSync () et commitAsync () . Comme son nom l'indique, commitSync() est un appel bloquantqui retourne après que les décalages ont été commitAsync() , alors que commitAsync() retourneimmédiatement. Si vous voulez savoir si une validation a réussi ou non, vous pouvez fournir ungestionnaire de OffsetCommitCallback ( OffsetCommitCallback ) à un paramètre de méthode. Faitesattention, dans les deux appels de validation, le consommateur valide les décalages du dernierappel poll() . Par exemple. supposons un sujet de partition unique avec un seul consommateur etle dernier appel à poll() renvoie des messages avec des décalages 4,5,6. Lors de la validation, ledécalage 6 sera validé car il s'agit du dernier décalage suivi par le client consommateur. En mêmetemps, commitSync() et commitAsync() permettent plus de contrôle sur le décalage que voussouhaitez valider: si vous utilisez les surcharges correspondantes vous permettant de spécifierune Map TopicPartition, OffsetAndMetadata le consommateur ne Map TopicPartition,OffsetAndMetadata que les décalages spécifiés (c.-à-d. que la carte peut contenir n'importe quelsous-ensemble de partitions assignées et que le décalage spécifié peut avoir n'importe quellevaleur).Sémantique des compensations engagéesUn décalage engagé indique que tous les messages jusqu'à ce décalage ont déjà été traités.Ainsi, comme les décalages sont des nombres consécutifs, la validation du décalage X valideimplicitement tous les décalages inférieurs à X Par conséquent, il n'est pas nécessaire de validerchaque décalage individuellement et de commettre plusieurs décalages à la fois, mais en validantle plus grand décalage.Faites attention, car de par sa conception, il est également possible de commettre un décalageplus petit que le dernier décalage engagé. Cela peut être fait si les messages doivent être lus uneseconde fois.Traitement des garantiesL'utilisation de la validation automatique fournit une sémantique de traitement au moins une fois.L'hypothèse sous-jacente est que poll() est uniquement appelé après que tous les messagesprécédemment livrés ont été traités avec succès. Cela garantit qu'aucun message ne soit perducar une validation se produit après le traitement. Si un consommateur tombe en panne avant unevalidation, tous les messages après la dernière validation sont reçus de Kafka et traités ànouveau. Cependant, cette tentative peut entraîner des doublons, car certains messages dudernier appel poll() ont peut-être été traités, mais l'échec s'est produit juste avant l'appel devalidation automatique.https://riptutorial.com/fr/home10

Si une sémantique de traitement au plus une fois est requise, la validation automatique doit êtredésactivée et un commitSync() manuel directement après poll() doit être effectué. Par la suite, lesmessages sont traités. Cela garantit que les messages sont validés avant leur traitement et nesont donc jamais lus une seconde fois. Bien sûr, certains messages peuvent être perdus en casd’échec.Comment puis-je lire le sujet depuis ses débutsIl existe plusieurs stratégies pour lire un sujet depuis ses débuts. Pour les expliquer, nous devonsd'abord comprendre ce qui se passe au démarrage du consommateur. Au démarrage d'unconsommateur, les événements suivants se produisent:1. rejoindre le groupe de consommateurs configuré, ce qui déclenche un rééquilibrage etattribue des partitions au consommateur2. rechercher les compensations validées (pour toutes les partitions affectées auconsommateur)3. pour toutes les partitions avec un offset valide, reprendre à partir de ce décalage4. pour toutes les partitions avec décalage non valide, définissez le décalage de départ enfonction du paramètre de configuration auto.offset.resetDémarrer un nouveau groupe de consommateursSi vous souhaitez traiter un sujet depuis son début, vous pouvez simplement démarrer unnouveau groupe de consommateurs (par exemple, choisir un group.id inutilisé) et définirauto.offset.reset earliest . Comme il n'y a pas de décalages validés pour un nouveau groupe,la réinitialisation automatique du décalage sera déclenchée et le sujet sera utilisé dès le début.Faites attention, au redémarrage du consommateur, si vous utilisez à nouveau le même group.id ,il ne lira pas le sujet de nouveau, mais reprendra où il est parti. Ainsi, pour cette stratégie, vousdevrez attribuer un nouveau group.id chaque fois que vous souhaitez lire un sujet depuis le début.Réutiliser le même identifiant de groupePour éviter de définir un nouveau group.id chaque fois que vous souhaitez lire un sujet depuis sondébut, vous pouvez désactiver la validation automatique (via enable.auto.commit false ) avant delancer le consommateur pour la toute première fois (en utilisant un group.id inutilisé group.id etparamètre auto.offset.reset earliest ). De plus, vous ne devez pas commettre de décalagemanuellement. Comme les décalages ne sont jamais validés avec cette stratégie, auredémarrage, le consommateur relira le sujet depuis le début.Cependant, cette stratégie présente deux inconvénients:1. ce n'est pas tolérant aux fautes2. le rééquilibrage du groupe ne fonctionne pas comme prévu(1) Comme les décalages ne sont jamais validés, un consommateur défaillant et arrêté est traitéde la même manière au redémarrage. Dans les deux cas, le sujet sera consommé dès le début.(2) Le décalage n'étant jamais validé, lors du rééquilibrage, les partitions nouvellement assignéeshttps://riptutorial.com/fr/home11

seront consommées dès le début.Par conséquent, cette stratégie ne fonctionne que pour les groupes de consommateurs avec unseul consommateur et ne devrait être utilisée qu'à des fins de développement.Réutiliser le même identifiant de groupe et la mêmevalidationSi vous souhaitez être tolérant aux pannes et / ou utiliser plusieurs consommateurs dans votregroupe de consommateurs, la validation des compensations est obligatoire. Ainsi, si voussouhaitez lire un sujet depuis le début, vous devez manipuler les décalages validés au démarragedu consommateur. Pour cela, KafkaConsumer fournit trois méthodes seek() , seekToBeginning() etseekToEnd() . Alors que seek() peut être utilisé pour définir un décalage arbitraire, les deuxième ettroisième méthodes peuvent être utilisées pour rechercher respectivement le début et la fin d’unepartition. Ainsi, en cas d'échec et de redémarrage, la recherche de redémarrage serait omise et leconsommateur pourrait reprendre sa sortie. Pour les utilisateurs de stop-and-restart-fromseekToBeginning() , seekToBeginning() serait appelé explicitement avant d'entrer dans votre bouclepoll() . Notez que seekXXX() ne peut être utilisé qu'après qu'un consommateur a rejoint un groupe- il est donc nécessaire d'effectuer un "sondage factice" avant d'utiliser seekXXX() . Le code globalserait quelque chose comme ceci:if (consumer-stop-and-restart-from-beginning) {consumer.poll(0); // dummy poll() to join consumer groupconsumer.seekToBeginning(.);}// now you can start your poll() loopwhile (isRunning) {for (ConsumerRecord record : consumer.poll(0)) {// process a record}}Lire Groupes de consommateurs et gestion des compensations en ensationshttps://riptutorial.com/fr/home12

Chapitre 3: les outils de la console kafkaIntroductionKafka propose des outils en ligne de commande pour gérer des sujets, des groupes deconsommateurs, pour consommer et publier des messages, etc.Important : les scripts de la console Kafka sont différents pour les plates-formes Unix etWindows. Dans les exemples, vous devrez peut-être ajouter l'extension en fonction de votre plateforme.Linux : scripts situés dans bin/ avec l'extension .sh .Windows : scripts situés dans bin\windows\ et avec l'extension .bat .Exampleskafka-topicsCet outil vous permet de lister, créer, modifier et décrire des sujets.Liste des sujets:kafka-topics--zookeeper localhost:2181 --listCréer un sujet:kafka-topicstopic test--create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --crée un sujet avec une partition et aucune réplication.Décrivez un sujet:kafka-topics--zookeeper localhost:2181 --describe --topic testModifier un sujet:# change configurationkafka-topics --zookeeper localhost:2181 --alter --topic test --configmax.message.bytes 128000# add a partitionkafka-topics --zookeeper localhost:2181 --alter --topic test --partitions 2(Attention: Kafka ne prend pas en charge la réduction du nombre de partitions d'un sujet) (voircette liste de propriétés de configuration )https://riptutorial.com/fr/home13

kafka-console-producteurCet outil vous permet de produire des messages à partir de la ligne de commande.Envoyer des messages de chaîne simples à un sujet:kafka-console-producer --broker-list localhost:9092 --topic testhere is a messagehere is another message D(chaque nouvelle ligne est un nouveau message, tapez ctrl D ou ctrl C pour arrêter)Envoyer des messages avec les clés:kafka-console-producer --broker-list localhost:9092 --topic test-topic \--property parse.key true \--property key.separator ,key 1, message 1key 2, message 2null, message 3 DEnvoyer des messages depuis un fichier:kafka-console-producer --broker-list localhost:9092 --topic test topic file.logkafka-console-consumerCet outil vous permet de consommer des messages d'un sujet.pour utiliser l'ancienne implémentation du consommateur, remplacez --bootstrapserver par --zookeeper .Afficher des messages simples:kafka-console-consumer --bootstrap-server localhost:9092 --topic testConsommez d'anciens messages:Pour voir les anciens messages, vous pouvez utiliser l'option --from-beginning .Afficher les messages de valeur-clé :kafka-console-consumer --bootstrap-server localhost:9092 --topic test-topic \--property print.key true \--property key.separator l.com/fr/home14

Ce consommateur est un outil de bas niveau qui vous permet de consommer des messages àpartir de partitions, de décalages et de répliques spécifiques.Paramètres utiles: : la partition spécifique à parition (par défaut à tous)offset : le décalage de début. Utilisez -2 pour consommer des messages depuis le début, -1pour consommer depuis la fin. max-messages : nombre de messages à imprimer replica : la réplique, par défaut pour le courtier leader (-1)paritionExemple:kafka-simple-consumer-shell \--broker-list localhost:9092 \--partition 1 \--offset 4 \--max-messages 3 \--topic test-topicaffiche 3 messages de la partition 1 commençant à l'offset 4 du sujet de test du sujet.kafka-groupes de consommateursCet outil vous permet de répertorier, décrire ou supprimer des groupes de consommateurs.Consultez cet article pour plus d'informations sur les groupes de consommateurs.Si vous utilisez toujours l'ancienne implémentation du consommateur, remplacez -bootstrap-server par --zookeeper .Liste des groupes de rap-server localhost:9092 --listDécrivez un groupe de consommateurs:kafka-consumer-groups opic01/127.0.0.1octopustest-topic12 /127.0.0.1localhost:9092 --describe --group octopusCURRENT-OFFSET LOG-END-OFFSET LAGOWNER15150octopus14151octopus-Remarques : dans la sortie ci-dessus, est le dernier décalage engagé de l'instance consommateur,log-end-offset est le log-end-offset le plus élevé de la partition (par conséquent, la sommede cette colonne vous donne le nombre total de messages pour le sujet) lag est la différence entre le décalage actuel du consommateur et le décalage le plus élevé,d'où la ome15

ownerest le client.id du consommateur (s'il n'est pas spécifié, un par défaut est affiché).Supprimer un groupe de consommateurs:la suppression n'est disponible que lorsque les métadonnées du groupe sont stockéesdans zookeeper (ancienne interface client). Avec la nouvelle API du consommateur, lecourtier gère tout, y compris la suppression des métadonnées: le groupe estautomatiquement supprimé lorsque le dernier décalage validé du groupe expire.kafka-consumer-groups --bootstrap-server localhost:9092 --delete --group octopusLire les outils de la console kafka en ligne: com/fr/home16

Chapitre 4: Producteur / Consommateur enJavaIntroductionCette rubrique montre comment produire et consommer des enregistrements en Java.ExamplesSimpleConsumer (Kafka 0.9.0)La version 0.9 de Kafka a introduit une refonte complète du consommateur de kafka. Si vous êtesintéressé par l'ancien SimpleConsumer (0.8.X), consultez cette page . Si votre installation de Kafkaest plus récente que la 0.8.X, les codes suivants devraient être intégrés.Configuration et initialisationKafka 0.9 ne prend plus en charge Java 6 ou Scala 2.9. Si vous êtes toujours sousJava 6, envisagez de passer à une version prise en charge.Tout d'abord, créez un projet Maven et ajoutez la dépendance suivante dans votre pom: dependencies dependency groupId org.apache.kafka /groupId artifactId kafka-clients /artifactId version 0.9.0.1 /version /dependency /dependencies Note : n'oubliez pas de mettre à jour le champ de version pour les dernières versions(maintenant 0.10).Le consommateur est initialisé à l'aide d'un objet Properties . Il y a beaucoup de propriétés vouspermettant d'affiner le comportement du consommateur. Voici la configuration minimale requise:Properties props new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "consumer-tutorial");props.put("key.deserializer", ue.deserializer", StringDeserializer.class.getName());Le bootstrap-servers est une liste initiale de courtiers permettant au consommateur de découvrir lereste du cluster. Cela ne doit pas nécessairement être tous les serveurs du cluster: le clientdéterminera l'ensemble complet des courtiers actifs parmi les courtiers de cette liste.https://riptutorial.com/fr/home17

Le deserializer indique au consommateur comment interpréter / désérialiser le

from: apache-kafka It is an unofficial and free apache-kafka ebook created for educational purposes. All the content is extracted from Stack Overflow Documentation, which is written by many hardworking individuals at Stack Overflow. It is neither