Kafka : Un système de messagerie distribué

  1. Présentation générale

Kafka est un système de file de messages distribué conçu par LinkedIn pour le traitement des journaux. LinkedIn génère de grandes quantités de données de journalisation, mais avec des exigences de fiabilité relativement faibles. Ces données incluent principalement les actions des utilisateurs (connexion, navigation, clics, partages, likes) et les journaux de fonctionnement du système (utilisation du CPU, mémoire, disque, réseau, état du système et des processus).

Le cluster Kafka se compose de plusieurs serveurs Broker. Chaque type de message est défini par un sujet (topic). Les messages d'un même sujet sont partitionnés (partition) selon une clé et un algorithme spécifiques, puis stockés sur différents Brokers. Les producteurs (producer) et les consommateurs (consumer) peuvent produire ou consommer des messages sur plusieurs Brokers pour un sujet donné.

  1. Concepts clés
  1. Les messages sont stockés sous forme de fichiers journaux. Les producteurs ne peuvent ajouter des messages qu'à la fin des fichiers existants, sans ID pour localiser les messages. Ils utilisent simplement le décalage (offset) dans le fichier pour identifier les messages. Ainsi, les consommateurs doivent lire les messages dans l'ordre du décalage, sans besoin d'une structure d'index complexe pour des accès aléatoires.
  2. Les Brokers ne gèrent pas ni ne coordonnent l'utilisation des messages par les utilisateurs. Les utilisateurs gèrent eux-mêmes leur décalage pour indexer les messages.
  3. L'unité minimale d'accès concurrentiel est la partition. Dans un groupe de consommateurs, tous les consommateurs (qui peuvent être considérés comme tous les processus en cours d'exécution d'une application) ne peuvent accéder qu'à une seule partition à la fois. Le nombre de partitions est fixe et ne peut pas être ajusté dynamiquement. Cela simplifie grandement le contrôle de concurrence entre les clients multiples/processus distribués, mais introduit également certaines restrictions sur les modes d'utilisation (par exemple, le niveau maximal de concurrence dépend du nombre prédéfini de partitions). De plus, les partitions entraînent un problème où les messages sont ordonnés uniquement au sein d'une partition et non globalement. Si une ordonnance globale est nécessaire, l'application doit utiliser d'autres mécanismes pour garantir cela.
  4. Kafka utilise un modèle de tirage (Pull) pour distribuer les messages. Le Broker ne suit pas ou ne maintient pas l'état d'utilisation des messages, tel que s'il reste des consommateurs qui n'ont pas lu un message ou s'il y a eu des lectures en double (en cours d'amélioration). La gestion des expirations des messages est simple et basée sur un minuteur qui supprime régulièrement les messages (par exemple, après 7 jours), ce qui simplifie les coûts de maintenance des messages.
  5. Depuis la version 0.8, Kafka a introduit un mécanisme de réplication des données. Plusieurs réplicas d'un même message sont répartis sur différents Brokers, avec un réplica leader responsable des lectures et écritures quotidiennes. Zookeeper supervise les réplicas followers. Les réplicas leaders sont équilibrés sur différents Brokers. Les producteurs peuvent choisir différents mécanismes d'acquittement (ack) : sans attendre l'acquittement du réplica leader, avec un acquittement partiel, ou après une sauvegarde complète. Ces mécanismes offrent des performances décroissantes (la débit du producteur peut diminuer de 1 à 3 fois) mais une robustesse des données croissante.
  6. Kafka optimise l'efficacité de transfert des données de diverses manières, notamment en permettant aux producteurs et consommateurs de lire/écrire des messages par lots pour réduire les coûts RPC, en utilisant Zero Copy pour transmettre directement le contenu des fichiers vers les sockets réseau au niveau du noyau, évitant ainsi la copie de données au niveau de l'application.
  7. Une gestion de la mémoire agressive, qui signifie essentiellement "ne pas gérer". Kafka ne gère pas de cache des messages à l'intérieur du processus JVM, lisant et écrivant directement depuis et vers les fichiers, en se basant sur le cache du système de fichiers au niveau du système d'exploitation. Cela évite les coûts supplémentaires des structures de données de gestion de cache dans la JVM et les coûts de performance liés au ramasse-miettes (GC). En maximisant l'utilisation du cache du système de fichiers et en évitant les coûts de performance relatifs aux lectures/écritures de fichiers par rapport à la lecture/écriture en mémoire, Kafka tire parti de ces optimisations.
  8. Stratégies de stockage
  1. Kafka gère les messages par sujet, chaque sujet contenant plusieurs partitions, chacune correspondant à un journal logique divisé en segments.
  2. Chaque segment stocke plusieurs messages. L'ID d'un message est déterminé par sa position logique, permettant de localiser directement le message sans mappage supplémentaire.
  3. Chaque partition est associée à un index en mémoire qui enregistre le décalage de la première message de chaque segment.
  4. Lorsqu'un message est publié sur un sujet, il est distribué uniformément sur plusieurs partitions (aléatoirement ou selon une fonction de rappel définie par l'utilisateur). Le Broker reçoit le message publié et l'ajoute à la fin du dernier segment de la partition correspondante. Lorsqu'un segment atteint un nombre de messages configuré ou un seuil de temps, ses messages sont vidangés sur le disque. Seuls les messages vidangés sur le disque peuvent être souscrits. Lorsqu'un segment atteint une certaine taille, il n'est plus écrit et le Broker crée un nouveau segment.
  5. Publication et abonnement aux messages
  1. Lors de la publication d'un message, le client Kafka construit un message, l'ajoute à un ensemble de messages (Kafka prend en charge la publication par lots, vous pouvez ajouter plusieurs messages à un ensemble et les publier en une seule fois), et spécifie le sujet du message lors de l'envoi.
  2. Pour s'abonner à un message, le client Kafka doit spécifier le sujet et le numéro de partition (chaque partition correspond à un flux de journal logique, par exemple, un sujet représente une ligne de produit, une partition représente les journaux de la ligne de produit segmentés par jour). Après l'abonnement, le client peut itérer pour lire les messages ; s'il n'y a pas de messages, le client bloque jusqu'à ce qu'un nouveau message soit publié.
  3. Un consommateur peut confirmer de manière cumulative les messages reçus. Lorsqu'il confirme un message à un décalage spécifique, cela signifie que tous les messages précédents ont été reçus avec succès, et le Broker met à jour l'enregistrement de décalage dans Zookeeper.
  4. Un sujet peut être consommé par plusieurs groupes de consommateurs distincts, mais chaque groupe de consommateurs ne peut consommer un message qu'une seule fois. Dans un groupe de consommateurs, chaque consommateur ne peut consommer que des partitions différentes. Autrement dit, une partition ne peut être consommée que par un seul consommateur.

5. Contrôle et coordination via Zookeeper

  1. Gestion de l'ajout et de la suppression dynamiques des Brokers et des consommateurs. Chaque Broker, lorsqu'il démarre, enregistre un nœud temporaire (enregistrement Broker) dans Zookeeper : contenant l'adresse IP et le port du Broker, les sujets et les informations des partitions qu'il stocke. Chaque consommateur, lorsqu'il démarre, enregistre également un nœud temporaire (enregistrement consommateur) dans Zookeeper : contenant le groupe de consommateurs auquel appartient le consommateur et les sujets qu'il souscrit.
  2. Déclenchement de l'équilibrage de charge. Lorsqu'un Broker ou un consommateur est ajouté ou supprimé, un algorithme d'équilibrage de charge est déclenché pour équilibrer la charge des souscriptions au sein d'un groupe de consommateurs.
  3. Maintenance des relations de consommation et des informations de chaque partition. Chaque groupe de consommateurs est associé à un enregistrement propriétaire temporaire et à un enregistrement de décalage persistant. Pour chaque partition souscrite, il existe un enregistrement propriétaire contenant l'ID du consommateur qui souscrit à cette partition ; il existe également un enregistrement de décalage contenant le décalage de la dernière souscription.

Résumé :

Kafka utilise Zookeeper pour stocker certaines informations métadonnées et utilise le mécanisme de surveillance (watch) de Zookeeper pour détecter les changements d'informations métadonnées et effectuer les actions appropriées (par exemple, équilibrage de charge lorsque un consommateur échoue).

  1. Enregistrement des nœuds Broker : Lorsqu'un Broker Kafka démarre, il enregistre d'abord ses informations de nœud dans Zookeeper (znode temporaire). Si la connexion entre le Broker et Zookeeper est interrompue, ce znode est également supprimé.

Format : /broker/ids/[0...N] --> hôte:port, où [0..N] représente l'ID du Broker (unique globalement), et la valeur de ce znode est l'information host:port de ce Broker.

  1. Enregistrement des sujets Broker : Lorsqu'un Broker démarre, il enregistre également les informations des sujets et des partitions qu'il détient dans Zookeeper, qui est également un znode temporaire.

Format : /broker/topics/[sujet]/[0...N], où [0..N] représente l'index de la partition.

  1. Consommateur et groupe de consommateurs : Chaque client consommateur est créé en enregistrant ses informations dans Zookeeper ; ce rôle est principalement destiné à l'"équilibrage de charge". Plusieurs consommateurs dans un groupe peuvent consommer de manière alternée toutes les partitions d'un sujet ; en d'autres termes, garantir que toutes les partitions de ce sujet peuvent être consommées par ce groupe, et pour des raisons de performance, permettre que les partitions soient réparties de manière équilibrée sur chaque consommateur.
  2. Enregistrement de l'ID du consommateur : Chaque consommateur a un ID unique (hôte:uuid, qui peut être spécifié via un fichier de configuration ou généré par le système), utilisé pour marquer les informations du consommateur,

Format : /consommateurs/[group_id]/ids/[consumer_id], qui est également un znode temporaire. La valeur de ce nœud est {"nom_du_sujet":#streams...}, c'est-à-dire indiquant le sujet + la liste des partitions actuellement consommées par ce consommateur.

  1. Suivi des décalages de consommation : Utilisé pour suivre le plus grand décalage consommé par chaque consommateur dans la partition qu'il consomme.

Format : /consommateurs/[group_id]/offsets/[sujet]/[broker_id-partition_id]-->valeur_offset, ce znode est un nœud persistant, ce qui montre que le décalage est lié au group_id, indiquant que lorsque un consommateur dans le groupe échoue, d'autres consommateurs peuvent continuer à consommer.

  1. Enregistrement du propriétaire de la partition : Utilisé pour marquer quelle partition est consommée par quel consommateur, znode temporaire

Format : /consommateurs/[group_id]/owners/[sujet]/[broker_id-partition_id] --> consumer_node_id

Lorsqu'un consommateur démarre, les opérations déclenchées sont :

A) Enregistrement d'abord dans l'"Enregistrement de l'ID du consommateur";

B) Ensuite, sous l'enregistrement de l'"Enregistrement de l'ID du consommateur", enregistrer une surveillance (watch) pour écouter la sortie ("leave") et l'arrivée ("join") d'autres consommateurs dans le même groupe ; toute modification de la liste de nœuds sous ce chemin de znode déclenchera un équilibrage de charge pour les consommateurs de ce groupe. (Par exemple, si un consommateur échoue, les autres consommateurs prennent en charge les partitions.)

C) Sous l'enregistrement du "Enregistrement des nœuds Broker", enregistrer une surveillance (watch) pour écouter la disponibilité des Brokers ; si la liste des Brokers change, cela déclenchera un équilibrage de charge pour tous les groupes de consommateurs.

En résumé :

  1. Le côté producteur utilise Zookeeper pour "découvrir" la liste des Brokers et établir des connexions socket avec les leaders de chaque partition d'un sujet pour envoyer des messages.
  2. Le côté Broker utilise Zookeeper pour enregistrer les informations Broker et surveiller la disponibilité des leaders de partition.
  3. Le côté consommateur utilise Zookeeper pour enregistrer les informations consommateur, y compris la liste des partitions consommées par le consommateur, et également pour découvrir la liste des Brokers et établir des connexions socket avec les leaders de partition pour obtenir des messages.
  4. Producteurs et consommateurs
  1. Producteurs

Équilibrage de charge : Le producteur maintient des connexions socket avec tous les leaders de partition d'un sujet ; les messages sont envoyés directement au Broker par le producteur, sans passer par une couche de routage. En fait, la partition vers laquelle un message est routé est décidée par le client producteur. Par exemple, on peut utiliser "aléatoire", "hachage de clé", "tourniquet", etc. Si un sujet contient plusieurs partitions, la réalisation d'une "distribution équilibrée des messages" au niveau du producteur est nécessaire. La position (host:port) des leaders de partition est enregistrée dans Zookeeper, et le producteur, en tant que client Zookeeper, a déjà enregistré une surveillance (watch) pour les événements de changement de leader de partition.

Envoi asynchrone : Les messages sont temporairement mis en tampon au niveau du client et envoyés par lots au Broker. Un grand nombre d'E/S de petits volumes ralentissent la latence globale du réseau. L'envoi différé par lots améliore en fait l'efficacité du réseau. Cependant, cela présente également certains risques, par exemple, lorsque le producteur échoue, les messages non encore envoyés seront perdus.

  1. Consommateurs

Le côté consommateur envoie une requête "fetch" au Broker et indique le décalage à partir duquel il souhaite obtenir des messages ; par la suite, le consommateur obtient un certain nombre de messages. Le côté consommateur peut également réinitialiser le décalage pour reconsommer des messages. Dans les implémentations JMS, le modèle Topic repose sur un mode push, où le Broker pousse les messages vers le côté consommateur. Cependant, dans Kafka, un mode pull est utilisé, où après avoir établi une connexion avec le Broker, le consommateur extrait (ou "fetch") activement les messages. Ce mode offre plusieurs avantages. Tout d'abord, le côté consommateur peut extraire et traiter les messages selon sa capacité de consommation, et contrôler le rythme de consommation des messages (décalage). De plus, le consommateur peut contrôler efficacement le nombre de messages consommés, extraire par lots.

Dans d'autres implémentations JMS, la position de la consommation des messages est conservée par le fournisseur pour éviter l'envoi répétitif de messages ou le renvoi de messages non consommés avec succès, tout en contrôlant l'état des messages. Cela nécessite beaucoup de travail supplémentaire du Broker JMS. Dans Kafka, un seul consommateur consomme les messages d'une partition, et il n'existe ni contrôle d'état des messages, ni mécanisme de confirmation de messages complexe, ce qui rend le Broker Kafka très léger. Une fois que le message a été reçu par le consommateur, celui-ci peut enregistrer localement le dernier décalage du message et enregistrer régulièrement le décalage dans Zookeeper. On voit donc que le côté client consommateur est également très léger.

  1. Mécanismes de transmission des messages Kafka dans le cas simple
  1. Au plus une fois : (le message est perdu si l'enregistrement du décalage dans Zookeeper réussit mais que le traitement du message échoue).
  2. Au moins une fois : (le message est envoyé au moins une fois si le traitement du message réussit mais que l'enregistrement du décalage dans Zookeeper échoue).
  3. Exactement une fois : (le message est envoyé exactement une fois si l'enregistrement du décalage et le traitement des données réussissent).

Au plus une fois : Le consommateur extrait (prend) le message, enregistre le décalage, puis traite le message. Si le client enregistre le décalage, mais qu'une exception se produit pendant le traitement du message, entraînant la non-traitation de certaines parties du message. Alors, ces messages "non traités" ne pourront plus être extraits, c'est ce qu'on appelle "au plus une fois".

Au moins une fois : Le consommateur extrait le message, traite le message, puis anregistre le décalage. Si le traitement du message réussit, mais que l'enregistrement du décalage échoue en raison d'une anomalie de Zookeeper, ce qui empêche l'exécution de l'opération d'enregistrement, cela entraîne le fait que lors de l'extraction suivante, il est possible d'obtenir à nouveau des messages déjà traités, c'est ce qu'on appelle "au moins une fois". La raison est que le décalage n'a pas été soumis à temps à Zookeeper, et lorsque Zookeeper est rétabli, il est toujours dans l'état précédent du décalage.

Exactement une fois : Kafka ne met pas en œuvre rigoureusement cette stratégie (basée sur une validation en deux phases, transactions), nous pensons que cette stratégie n'est pas nécessaire dans Kafka.

Généralement, "au moins une fois" est notre choix préféré. (Comparé à "au plus une fois", recevoir des données en double est préférable à en perdre).

  1. Réplication et sauvegarde

La réplication est maître-esclave, le leader suit l'état des followers, et les retire si leur performance est insuffisante, garantissant qu'au moins un est en bon état. Ainsi, lorsque le producteur sauvegarde des données, le succès est considéré lorsque le leader et les followers ont réussi. Si le leader échoue, un nouveau leader est élu parmi les followers, généralement celui ayant les données les plus complètes. Cependant, il faut aussi prendre en compte le nombre de leaders sur ce broker.

Kafka réplique les données de chaque partition sur plusieurs serveurs, avec un leader et des followers (qui peuvent être absents). Le nombre de réplicas peut être défini dans le fichier de configuration du broker. Le leader gère toutes les requêtes de lecture et d'écriture, tandis que les followers doivent rester synchronisés avec le leader. Les followers consomment les messages et les sauvegardent localement dans leurs journaux, tout comme les consommateurs. Le leader suit l'état de tous les followers, et si un follower "tombe en retard" ou échoue, le leader le supprime de la liste des réplicas synchronisés. Un message est considéré comme "validé" lorsque tous les followers l'ont sauvegardé avec succès, et seulement alors le consommateur peut le consommer. Même si un seul réplica est en vie, la transmission et la réception des messages peuvent être garanties, tant que le cluster Zookeeper est actif. (Contrairement à d'autres systèmes de stockage distribués, comme HBase, qui nécessitent la survie d'une majorité.)

Lorsqu'un leader échoue, il faut élire un nouveau leader parmi les followers. Il est possible que les followers soient en retard par rapport au leader, donc il faut choisir un follower "à jour". Lors de l'élection du nouveau leader, il faut également prendre en compte le nombre de leaders de partition déjà présents sur le serveur. Si un serveur a trop de leaders de partition, cela signifie qu'il subit davantage de pression I/O. Lors de l'élection du nouveau leader, il faut tenir compte de l'équilibrage de charge.

  1. Format des fichiers journaux

Si le nom du sujet est "mon_sujet" et qu'il comporte 2 partitions, les journaux seront enregistrés dans les répertoires mon_sujet_0 et mon_sujet_1. Les fichiers journaux contiennent une série d'entrées de journal (log entries), chaque entrée ayant le format "4 octets représentant la longueur N du message" + "N octets de contenu du message". Chaque journal a un décalage qui identifie de manière unique un message, la valeur du décalage étant un nombre sur 8 octets représentant la position de départ du message dans cette partition. Physiquement, chaque partition est composée de plusieurs fichiers journaux (appelés segments). Les fichiers segments sont nommés "décalage_minimum".kafka, par exemple "00000000000.kafka". Le "décalage_minimum" indique le décalage du premier message dans ce segment.

Remarques :

  1. La liste des segments détenu par chaque partition est stockée dans Zookeeper.
  2. Lorsqu'un fichier segment atteint une certaine taille (configurable, par défaut 1 Go), un nouveau fichier est créé.
  3. Lorsque le nombre de messages dans le tampon atteint une certaine valeur (ou que le temps écoulé depuis la dernière vidange atteint un certain seuil), une vidange des informations du journal vers le fichier journal est déclenchée.
  4. Vérification et réparation : Si un Broker échoue, il est possible de perdre les messages qui n'ont pas encore été vidangés sur le disque. En raison d'une panne du serveur, la structure du fichier journal peut être endommagée (à la fin du fichier), il est donc nécessaire de vérifier la validité de la structure du dernier segment de fichier lors du démarrage du serveur et d'effectuer les réparations nécessaires.
  5. Pour obtenir un message, il faut spécifier le décalage et la taille maximale du bloc : le décalage indique la position de départ du message, et la taille maximale du bloc indique la longueur totale maximale des messages à obtenir (indiquant indirectement le nombre de messages). Selon le décalage, on trouve le fichier segment contenant le message, puis on calcule la différence entre le décalage minimum du segment et le décalage pour obtenir sa position relative dans le fichier, puis on lit et on sort directement.
  6. La stratégie de suppression des fichiers journaux est très simple : un thread en arrière-plan scanne régulièrement la liste des fichiers journaux et supprime ceux dont la durée de conservation dépasse un certain seuil (selon la date de création du fichier). Pour éviter la suppression d'un fichier lorsqu'il est toujours lu (consommation par un consommateur), une méthode copy-on-write est utilisée (lors de la copie d'un objet, l'objet original n'est pas réellement copié à une autre position en mémoire, mais un pointeur est défini dans la table de mappage mémoire de l'objet nouveau, pointant vers la position de l'objet source, et le bit Copy-On-Write de cette zone mémoire est défini à 1. Ainsi, lorsqu'une opération de lecture est exécutée sur le nouvel objet, les données mémoire ne changent pas et l'opération de lecture est effectuée directement ; lorsqu'une opération d'écriture est exécutée sur le nouvel objet, l'objet original est réellement copié à une nouvelle adresse mémoire, la table de mappage mémoire de l'objet nouveau est modifiée pour pointer vers cette nouvelle position, et l'opération d'écriture est effectuée sur cette nouvelle position.).
  7. Configuration principale
  1. Configuration du Broker
 1 broker.id=1
 2 port=9091
 3 num.network.threads=2
 4 num.io.threads=2
 5 socket.send.buffer.bytes=1048576
 6 socket.receive.buffer.bytes=1048576
 7 socket.request.max.bytes=104857600
 8 log.dir=./logs
 9 num.partitions=2
10 log.flush.interval.messages=10000
11 log.flush.interval.ms=1000
12 log.retention.hours=168
13 #log.retention.bytes=1073741824
14 log.segment.bytes=536870912
15 num.replica.fetchers=2
16 log.cleanup.interval.mins=10
17 zookeeper.connect=192.168.0.1:2181,192.168.0.2:2182,192.168.0.3:2183
18 zookeeper.connection.timeout.ms=1000000
19 kafka.metrics.polling.interval.secs=5
20 kafka.metrics.reporters=kafka.metrics.KafkaCSVMetricsReporter
21 kafka.csv.metrics.dir=/tmp/kafka_metrics
22 kafka.csv.metrics.reporter.enabled=false

2. Configuration principale du consommateur

3. Configuration principale du processeur

Notes supplémentaires :

1. public Map<String, List<KafkaStream<byte[], byte[]>>> createMessageStreams(Map<String, Integer> topicCountMap), où la clé de la carte est le nom du sujet, et la valeur est le nombre de partitions pour ce sujet. Par exemple, si le sujet n'existe pas dans Kafka, il sera créé avec le nombre de partitions spécifié par la valeur. Si le sujet existe déjà, cette valeur n'affecte pas la configuration.


2. Pour envoyer des données vers une partition spécifique, il faut définir la classe partitionner par le biais de l'attribut partitioner.class. Si une partition spécifique est définie manuellement, il faut écrire le programme correspondant. Par défaut, c'est kafka.producer.DefaultPartitioner, qui utilise un hachage basé sur la clé.
  1. Pour garantir l'unicité des données lues par plusieurs consommateurs d'un même sujet, il faut définir la même valeur pour le group_id de ces consommateurs, ce qui crée une structure de file de messages similaire à une queue. Si des valeurs différentes sont définies, cela ressemble à une structure de diffusion.
  2. Dans l'API consommateur, les paramètres impliquant des nombres, tels que Map, numStream, font référence à la création d'un sujet lorsqu'il n'existe pas, avec un nombre de partitions Integer.numStream. Notez que si ce nombre est supérieur à la propriété num.partitions configurée dans le Broker, le nombre de partitions créées sera basé sur la propriété num.partitions.
  3. Dans l'API producteur, lors de l'appel de send, si le sujet n'existe pas, il sera créé. Cette méthode ne fournit pas de paramètre pour le nombre de partitions, donc le nombre de partitions est déterminé par la propriété num.partitions configurée dans le Broker.

Les résumés ci-dessus proviennent de Transwisenet, merci au professeur Wang Senfeng de Transwisenet.

Étiquettes: Kafka Apache Kafka messages distribués file de messages consommateur Kafka

Publié le 8 août à 14h01

Étiquettes Populaires

©2026 L'Atelier Monstre