Kafka 根据请求使用单个消息

Question

我想根据每个请求使用 gunicorn 制作一个烧瓶应用程序/API-

1. 从 Kafka 主题中读取单个值
2. 做一些处理
3. 并将处理后的值返回给用户（或任何调用 API 的应用程序）。

所以，到目前为止，我找不到任何例子。那么，以下函数是正确的方法吗？

consumer = KafkaConsumer(
        "first_topic",
        bootstrap_servers='xxxxxx',
        auto_offset_reset='xxxx',
        group_id="my_group")

def get_value_from_topic:
    for msg in consumer:
        return msg

if __name__ == "__main__":
    print(get_value_from_topic())

或者有没有更好的方法来使用像 Faust 这样的库？ 我使用 Kafka 的原因是为了避免烧瓶工作人员之间同步的所有麻烦（在传统数据库的情况下），因为我只想使用一次来自 Kafka 的值。

Answer 1

乍一看，这似乎没问题。您的消费者迭代器被迭代一次，然后您返回该值。

不过，更惯用的方法是这样

def get_value_from_topic():
    return next(consumer)

但是，使用您的其他设置，不能保证这只会轮询一条消息，因为 Kafka 消费者分批轮询，并且会自动提交这些批次的偏移量。因此，您需要禁用自动提交并自行处理，例如在处理 http 请求后提交将给您至少一次交付，而在之前提交将给您最多一次。由于您正在与 HTTP 服务器交互，因此 Kafka 无法准确地为您提供一次处理