Kafka - Retard dans la File d'attente de la mise en œuvre de haut niveau de consommation

Une façon de le faire serait d'utiliser un autre sujet où vous poussez tous les messages qui doivent être retardée. Si tous les messages différés doivent être traitées après le même temps de retard, ce sera assez simple:

while(it.hasNext()) {
    val message = it.next().message()

    if(shouldBeDelayed(message)) {
        val delay = 24 hours
        val delayTo = getCurrentTime() + delay
        putMessageOnDelayedQueue(message, delay, delayTo)
    }
    else {
       process(message)
    }

    consumer.commitOffset()
}

Tous les messages réguliers seront désormais traités dès que possible, tandis que ceux qui a besoin d'un délai est mis sur un autre sujet.

La bonne chose est que nous savons que le message à la tête de la le retard sujet est celui qui doit être traitée en premier depuis sa delayTo valeur sera le plus petit. Par conséquent, nous pouvons mettre en place un autre consommateur qui lit la tête de message, vérifie si le timestamp est dans le passé et, dans l'affirmative traite le message et s'engage à les compenser. Si non, elle n'engage pas le décalage et la place dort jusqu'à ce que le temps:

while(it.hasNext()) {
    val delayedMessage = it.peek().message()
    if(delayedMessage.delayTo < getCurrentTime()) {
        val readMessage = it.next().message
        process(readMessage.originalMessage)
        consumer.commitOffset()
    } else {
        delayProcessingUntil(delayedMessage.delayTo)
    }
}

Dans le cas où il y a de diffrent temps de retard, vous pouvez partitionner le sujet sur le retard (par exemple, 24 heures, 12 heures, 6 heures). Si le temps de retard est plus dynamique que celui qu'elle devient un peu plus complexe. Vous pourriez résoudre par l'introduction d'avoir deux de retard sujets. Lire tous les messages en retard à la coupure sujet A et de traiter tous les messages dont delayTo valeur sont dans le passé. Parmi les autres que tu viens de trouver celui le plus proche de delayTo et ensuite de les placer sur le sujet B. Le sommeil jusqu'à ce que le plus proche doit être traitée de tout faire dans le sens inverse, c'est à dire de traiter les messages du sujet B et de mettre à la fois à ne pas encore être traité correctement de retour sur le sujet A.

Pour répondre à vos questions (certains ont été abordés dans les commentaires à votre question)

1. commettre chaque décalage peut ralentir ZK bas

Vous pourriez envisager de passer à la mémorisation de l'offset dans Kafka (une fonctionnalité disponible à partir de 0.8.2, découvrez offsets.storage des biens de consommation config)

1. du consommateur.commitOffsets lever une exception? si oui, je vais consommer le même message deux fois (peut résoudre avec la quantité de messages)

Je crois qu'il peut s'il n'est pas en mesure de communiquer avec le décalage de stockage par exemple. À l'aide de la quantité de messages permet de résoudre ce problème tu, comme tu dis.

1. problème d'attente long temps sans commettre le décalage, par exemple le délai est de 24 heures, sera à côté de itérateur, le sommeil pendant 24 heures, de processus et de s'engager (ZK délai d'expiration de session ?)

Ce ne sera pas un problème avec la solution décrite ci-dessus, à moins que le traitement du message lui-même prend de plus que le délai d'expiration de session.

1. comment peut-ZK session keep-alive sans commettre de nouvelles compensations ? (définition d'une ruche zookeeper.session.délai d'attente.ms peut résoudre les morts à la consommation sans le reconnaître)

À nouveau avec ce qui précède, vous ne devriez pas avoir besoin de définir une longue expiration de la session.

1. d'autres problèmes im manque?

Il en a toujours 😉

OriginalL'auteur Emil H

Utilisation Tibco EMS ou d'autres files d'Attente JMS. Ils ont réessayer retard construit en . Kafka n'est pas le bon choix de conception pour ce que vous faites

OriginalL'auteur Dhyan