Kafka-使用高级使用者的延迟队列实施

小编典典

Kafka-使用高级使用者的延迟队列实施

java

想要使用高级消费者API实现延迟的消费者

大意：

按键生成消息（每个消息包含创建时间戳记），以确保每个分区按生成时间对消息进行排序。
auto.commit.enable = false（将在每个消息处理之后显式提交）
消费一条消息
检查消息时间戳，并检查是否经过了足够的时间
处理消息（此操作将永不失败）
提交1个偏移

        while (it.hasNext()) {
      val msg = it.next().message()
      //checks timestamp in msg to see delay period exceeded
      while (!delayedPeriodPassed(msg)) { 
         waitSomeTime() //Thread.sleep or something....
      }
      //certain that the msg was delayed and can now be handled
      Try { process(msg) } //the msg process will never fail the consumer
      consumer.commitOffsets //commit each msg
    }

有关此实现的一些担忧：

提交每个偏移量可能会使ZK变慢
Consumer.commitOffsets是否可以引发异常？如果是，我将使用同一条消息两次（可以解决幂等消息）
等待较长时间而不提交偏移量的问题，例如延迟时间为24小时，将从迭代器中获取下一个，睡眠24小时，进行处理并提交（ZK会话超时？）
ZK会话如何在不提交新偏移的情况下保持活动？（设置一个配置单元zookeeper.session.timeout.ms可以解决死掉的消费者而又不认识它）
我还有其他问题吗？

谢谢！

阅读 1595

2020-11-13

共1个答案

小编典典

解决此问题的一种方法是使用不同的主题，在其中推送所有将要延迟的消息。如果所有延迟的消息都应在相同的时间延迟之后进行处理，这将很简单：

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()
}

现在将尽快处理所有常规消息，而需要延迟的消息将放在另一个主题上。

令人高兴的是，我们知道延迟主题开头的消息是应该首先处理的消息，因为其delayTo值将是最小的。因此，我们可以设置另一个读取头消息的使用者，检查时间戳是否在过去，如果是，则处理该消息并提交偏移量。如果不是，它不会提交偏移量，而是一直休眠直到那个时间：

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)
    }
}

如果存在不同的延迟时间，则可以按延迟划分主题（例如24小时，12小时，6小时）。如果延迟时间比该时间更动态，那么它将变得更加复杂。您可以通过引入两个延迟主题来解决。从延迟主题中读取所有消息，A并处理所有delayTo值都为过去的消息。除其他外，您只需找到最接近的一个，delayTo然后将它们放在主题上即可B。休眠直到应该处理最接近的消息为止，然后相反地进行所有处理，即处理来自主题的消息，B然后将尚未处理的消息放回主题A。

回答您的特定问题（在您的问题的注释中已经解决了一些问题）

提交每个偏移量可能会使ZK变慢

您可以考虑切换到在Kafka中存储偏移量（自0.8.2起可用的功能，请offsets.storage在消费者配置中检出属性）

Consumer.commitOffsets是否可以引发异常？如果是，我将使用同一条消息两次（可以解决幂等消息）

我认为，例如，如果它不能与偏移存储进行通信，则可以。正如您所说，使用幂等消息可以解决此问题。

等待较长时间而不提交偏移量的问题，例如延迟时间为24小时，将从迭代器中获取下一个，睡眠24小时，进行处理并提交（ZK会话超时？）

除非消息本身的处理花费的时间超过会话超时，否则上述解决方案不会有问题。

4.
ZK会话如何在不提交新偏移的情况下保持活动？（设置一个配置单元zookeeper.session.timeout.ms可以解决死掉的消费者而又不认识它）

同样，使用上述方法，您无需设置长时间的会话超时。

我还有其他问题吗？

总是有;）

2020-11-13