对于事件驱动的kafka消费者有pythonapi吗?

eh57zj3b  于 2021-06-06  发布在  Kafka
关注(0)|答案(2)|浏览(424)

我一直在尝试建立一个 flask 应用程序,Kafka作为唯一的接口。因此,我希望有一个kafka消费者,当相关主题的流中有新消息时触发,并通过将消息推回到kafka流来响应。
我一直在寻找类似spring的实现:

@KafkaListener(topics = "mytopic", groupId = "mygroup")
public void listen(String message) {
    System.out.println("Received Messasge in group mygroup: " + message);
}

我看过:
KafkaPython
皮Kafka
合流Kafka
但我在python中找不到任何与事件驱动的实现风格相关的东西。

qzlgjiam

qzlgjiam1#

Kafka消费者必须不断地投票,从经纪人那里检索数据。
spring为您提供了这个奇特的api,但实际上,它在一个循环中调用poll,并且只在检索到记录时调用您的方法。
您可以轻松地构建与前面提到的任何python客户机类似的东西。与java一样,这不是(大多数)kafka客户机直接公开的api,而是由顶层提供的api。这是你需要建立的东西。

rqqzpn5f

rqqzpn5f2#

下面是@mickaelmaison的答案给出的想法的实现。我用的是KafkaPython。

from kafka import KafkaConsumer
import threading

BOOTSTRAP_SERVERS = ['localhost:9092']

def register_kafka_listener(topic, listener):

# Poll kafka

    def poll():
        # Initialize consumer Instance
        consumer = KafkaConsumer(topic, bootstrap_servers=BOOTSTRAP_SERVERS)

        print("About to start polling for topic:", topic)
        consumer.poll(timeout_ms=6000)
        print("Started Polling for topic:", topic)
        for msg in consumer:
            print("Entered the loop\nKey: ",msg.key," Value:", msg.value)
            kafka_listener(msg)
    print("About to register listener to topic:", topic)
    t1 = threading.Thread(target=poll)
    t1.start()
    print("started a background thread")

def kafka_listener(data):
    print("Image Ratings:\n", data.value.decode("utf-8"))

register_kafka_listener('topic1', kafka_listener)

轮询是在另一个线程中完成的。一旦收到消息,就会通过传递从kafka检索到的数据来调用侦听器。

相关问题