我的制作人 n 来自单个输入消息的任务，并在上发布这些任务 topic .
要求是 topic ，其中任何一个进程都不应超过3个 n 1小时内完成任务。
这意味着如果我想立即处理所有这些消息，我至少需要 ceil(n/3) 消费者。如果少于 ceil(n/3) 那么，我需要一些方法来推迟一个消息，直到 num_processed < 3 在最后一个小时。
就实现这个解决方案的实用性而言，我希望将Kafka与《浮士德》[1]结合使用，但如果有必要，我也可以访问redis。
到目前为止，我的想法是确保至少 ceil(n/3) 消费者在生产时，然后只是使用循环分配的任务 topic 制作人。无论如何，这是最佳的解决方案，因为它避免了需要等待1小时来处理消息。然而，这只会工作，直到足够的消费者死亡，因此，超过3个可以由同一消费者最有可能在1小时内处理。这是不可接受的。
另一个想法可能是让消费者在每次接收消息时检查是否执行了其中的3个 n 任务已经完成了，如果是这样的话，不知何故请求另一个使用者处理它——但是我在Kafka中找不到任何合适的机制来实现这一点。
[1] https://faust.readthedocs.io/
1条答案
按热度按时间mgdq6dx11#
有些东西需要更多的努力，但会使实际的处理容易得多，是有一个预消费者，只需等待，直到有3个消息消费，打包成一个“元消息”，并发送到一个“准备处理”的主题。它必须像@cricket\u007提到的那样，在它实际消耗了3条消息并将它们生成出站主题之前，它不应该提交。
这样，最终的消费者就是一个极其简单的消费者。它只需要从“ready for processing”主题中消费，只要它收到一条消息，您就知道它将包含您需要的3个事件。您只需处理它们，然后再等一个小时，直到您可以再次投票。没有必要与任何其他消费者进行协调。