Nous avons une implémentation rabbitmq qui fonctionne, en raison du volume, nous prévoyons de passer à kafka.
J'ai un doute à un moment donné.
Dans rabbitMQ, lorsque le consommateur consomme le message du Q, le message passe à une étape différente, non acquittée. le client / consommateur prend un certain temps pour traiter le message, une fois le processus réussi, il envoie un accusé de réception au Q et le message est supprimé du Q. en cas d'échec, après une période définie si le Q n'obtient pas d'accusé de réception, le message est annexé à la fin du Q. De cette façon, nous ne perdons aucun message.
Avec mon peu de connaissances en Kafka, je comprends que si, par exemple, le message 100 n'a pas été traité avec succès, le décalage n'a pas été augmenté, mais il sera augmenté si le message 101 est traité avec succès. J'ai donc perdu le message 100.
Existe-t-il un moyen de garantir qu'aucun des messages ne sera perdu.
4 Réponses :
Le décalage de votre message ne sera pas augmenté à moins que vous n'interrogiez de nouveaux messages. Vous devez donc vous préoccuper du retraitement de votre message.
Si vous souhaitez stocker le résultat de votre traitement de données dans le cluster Kafka, vous pouvez utiliser fonction de transaction de Kafka . De cette façon, vous pouvez prendre en charge une livraison unique. Toutes vos modifications seront enregistrées ou aucune d’entre elles ne sera stockée.
Une autre approche consiste à rendre votre scénario de traitement idempotent. Vous attribuerez un identifiant unique à chaque message dans Kafka. Lorsque vous traitez le message, vous stockez l'ID dans une base de données. Après un plantage, vous vérifiez que votre identifiant de message est déjà traité en consultant la base de données.
Kafka ne supprime pas les messages des sujets à moins qu'il n'atteigne l'un des log.retention.bytes log.retention.hours log.retention.minutes < / code> log.retention.ms configs. Ainsi, si le décalage augmente, vous ne perdez pas les messages précédents et vous pouvez simplement modifier le décalage à la position souhaitée.
@ admin, mon consommateur ne gardait pas la trace, et nous avons des centaines de clients, qui consomment des messages RabbitMQ. Ils consomment et reconnaissent chaque message. et RabitMQ s'occupe de supprimer un message consommé avec succès, et de le publier à nouveau sur Q pour les messages qui ont échoué. Maintenant, si le décalage a avancé pour un client sans traiter un message, comment le client consomme à nouveau le même message, sans changer à la fin des clients.
Vous devriez lire un peu comment fonctionne la consommation de messages dans Kafka. Voici un lien vers la section consommateur de la documentation officielle de Kafka: https://kafka.apache.org/ documentation / # theconsumer
Fondamentalement, dans Kafka, les messages ne sont supprimés qu'après un temps suffisant, et cela est configuré à l'aide de log.retention.hours , log.retention.minutes et log.retention.ms comme @Amin l'a dit.
Dans Kafka, n'importe quel nombre de consommateurs peut commencer à consommer des messages à partir de n'importe quel sujet, à tout moment, que d'autres consommateurs consomment déjà ou non à partir de ce même sujet. Kafka garde une trace de l'emplacement de chaque consommateur, sur chaque sujet / partition, à l'aide de décalages stockés dans Kafka lui-même. Ainsi, si votre consommateur a besoin de consommer le message 100, comme vous l'avez décrit dans votre question, vous pouvez simplement "revenir en arrière" vers le message souhaité, et recommencer à consommer normalement. Peu importe si vous l'avez déjà consommé ou si d'autres consommateurs lisent ou non ce sujet.
À partir de la documentation officielle de Kafka:
Un consommateur peut délibérément revenir à un ancien décalage et re-consommer les données. Cela viole le contrat commun d'une file d'attente, mais s'avère être une caractéristique essentielle pour de nombreux consommateurs. Par exemple, si le code consommateur a un bug et est découvert après quelques messages sont consommés, le consommateur peut re-consommer ces messages une fois le bogue est corrigé.
Merci @mjuarez, Mon scénario est un peu différent, laissez-moi vous expliquer, donc je n'ai qu'un seul consommateur / lecteur par sujet, si j'en ai plusieurs, ils sont pour le parallélisme, donc ils liront / consommeront des messages différents. Mais le consommateur / lecteur qui ne parvient pas à traiter un message, et le transmet au suivant, peut être que le message restera toujours dans kafka, mais il continuera d'augmenter le décalage, ne lisant que les nouveaux messages.Donc celui qui a échoué ne sera pas traité encore . Pour rabbitMq, c'était facile, car il ramènera le message vers le Q. Kafka conserve un offset par consommateur, RabbitMQ par message.
Configurez simplement votre client pour qu'il lise les messages kafka 1 à la fois et ne validez les offsets qu'après avoir traité avec succès chaque message, puis vous aurez un comportement similaire à rabbitmq (seulement plus rapidement lorsque vous utilisez des partitions de rubrique pour le partage et plusieurs consommateurs dans un groupe de consommateurs commun ) Il sera plus lent que le mode normal où les clients Kafka consomment les messages par lots mais toujours plus rapide
De plus, les clients kafka conservent généralement un décalage par partition par groupe de consommateurs, de sorte qu'une rubrique avec 16 partitions et 4 consommateurs dans un seul groupe de consommateurs conserverait 16 décalages (un par partition).
@Hans c'est une solution qui fonctionne, mais qui finira par ralentir le processus. Y a-t-il une alternative, sinon nous devons nous en tenir à l'ancien RabbitMq
L'alternative est de rechercher des lots de messages Kafka et de gérer vos propres compensations. Si vous devez ignorer un message, vous écrivez ses décalages dans une file d'attente de messages morts ou dans un système externe non Kafka pour un retraitement ultérieur.
J'ai également été confronté à la même question. Si je veux mettre de manière simple, RabbitMQ tient un compte de chacun
Kafka ne le fait pas, donc vous ne pouvez pas l'avoir prêt, vous devez l'implémenter vous-même.
Il existe cependant des options disponibles, utilisez kmq, les performances deviendront inférieures à 50%, jetez un œil
https://softwaremill.com/kafka-with-selective-acknowledgments- performances /
Vous devrez implémenter une file d'attente de lettres mortes (DLQ).
Pourquoi ne pas considérer Solace comme un substitut de messagerie plus proche de Rabbit? API similaires + sémantique (par exemple, ACK par message), mais meilleure gestion des volumes.