如何批量消费来自kafka生产者的消息（kafka-python）

Question

我在 python 中有一个 kafka 生产者和消费者。我希望批量消费来自 kafka 生产者的消息，比如说 2。从生产者那里，我一直在发送如下电子邮件数据：

[{
    "email" : "sukhi215c@gmail.com",
    "subject": "Test 1",
    "message" : "this is a test"
},
{
    "email" : "sukhi215c@gmail.com",
    "subject": "Test 2",
    "message" : "this is a test"   
},
{
    "email" : "sukhi215c@gmail.com",
    "subject": "Test 3",
    "message" : "this is a test"   
},
{
    "email" : "sukhi215c@gmail.com",
    "subject": "Test 4",
    "message" : "this is a test"   
}]

我正在尝试批量使用这些数据。我希望一次使用 2 条消息并根据这 2 条数据发送电子邮件并使用下一组数据。我尝试的解决方法是：

consumer = KafkaConsumer(topic, bootstrap_servers=[server], api_version=(0, 10))
for message in consumer[:2]:
    string = message.value.decode("utf-8")
    dict_value = ast.literal_eval(string)

我得到的错误是：

    for message in consumer[:2]:
TypeError: 'KafkaConsumer' object is not subscriptable

有人可以帮我解决这个问题吗？

Answer 1

消费者不是一个集合；它的迭代器是无限的。

如果您想每两个事件执行一次操作，请使用计数器或您自己的列表

data = []
consumer = KafkaConsumer(topic, bootstrap_servers=[server], api_version=(0, 10))
for message in consumer:
    data.append(message)
    if len(data) >= 2:
        action(data)
        data.clear()