在kafka（或其他消息队列）中是否有一个功能，通过它我们可以添加只在指定时间内处于活动状态的消息。这意味着消息不应该在指定的时间之前被使用。接下来，是否有一种方法可以在到达预定时间时将这些消息移到队列的前面。
pftdvrlh1#
不在kafka代理本身中，但您可以通过定义自定义处理器来使用kafka流执行此操作，该处理器执行以下操作：将每条消息存储到statestore中，使用消息处理时间戳作为密钥。由于多条消息可能具有相同的时间戳，因此需要将它们 Package 到一个集合中，或者在键的后面加上一些唯一的值。在init方法中，调度一个标点符号，它周期性地对状态存储进行范围查询，转发到达其时间戳的消息，并将它们从存储中删除。转发的消息将进入一个单独的主题，消费者将不得不从该主题中消费，但那里的消息将按照您定义的顺序。
3mpgtkmj2#
Kafka不能这样做。Kafka基本上只是一个提交日志，它将按照接收消息的顺序写入消息，并按照相同的顺序读取消息。但是其他更传统的消息代理可以做到这一点，例如activemqartemis支持调度消息。我不认为消息会在队列中移动，但它不会在计划的时间之前被消耗。
2条答案
按热度按时间pftdvrlh1#
不在kafka代理本身中，但您可以通过定义自定义处理器来使用kafka流执行此操作，该处理器执行以下操作：
将每条消息存储到statestore中，使用消息处理时间戳作为密钥。由于多条消息可能具有相同的时间戳，因此需要将它们 Package 到一个集合中，或者在键的后面加上一些唯一的值。
在init方法中，调度一个标点符号，它周期性地对状态存储进行范围查询，转发到达其时间戳的消息，并将它们从存储中删除。
转发的消息将进入一个单独的主题，消费者将不得不从该主题中消费，但那里的消息将按照您定义的顺序。
3mpgtkmj2#
Kafka不能这样做。Kafka基本上只是一个提交日志，它将按照接收消息的顺序写入消息，并按照相同的顺序读取消息。
但是其他更传统的消息代理可以做到这一点，例如activemqartemis支持调度消息。我不认为消息会在队列中移动，但它不会在计划的时间之前被消耗。