Skip to main content
 首页 » 编程设计

python之在kafka-python的客户端中消费消息

2024年10月24日31myhome

我是卡夫卡的新手。通过一些 kafka-python 在线教程的帮助,我编写了以下代码:

   from kafka import SimpleProducer, KafkaClient, KafkaConsumer 
   kafka =  KafkaClient("localhost:9092") 
   producer = SimpleProducer(kafka) 
   producer.send_messages(b'my-topic', b'this method', b'Hello World') 
   consumer = KafkaConsumer('my-topic', 
                     group_id='my_group', 
                     bootstrap_servers=['localhost:9092']) 
   for message in consumer: 
       print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, 
                                         message.offset, message.key, 
                                         message.value)) 

但问题是在最后一个 for 循环中,代码执行被卡住了,我无法弄清楚。

请您参考如下方法:

您的原始代码是正确的。 我运行您的代码。 KafkaConsumer 可以用来消费消息。如果您打开另一个控制台,并运行相同的原始代码,您将看到输出。

在http://kafka-python.readthedocs.org/en/latest/apidoc/kafka.consumer.html ,有很多Consumer类,比如SimpleConsumer、KafkaConsumer、Consumer等等。为什么你的代码卡在 for 循环中?因为你代码中的consumer默认设置为消费new消息。在这种情况下,消费者不会使用 producer.send_messages() 函数生成的消息。

顺便说一下,如果您使用 SimpleConsumer,您可以使用 seek() 来设置您想要消费的消息的偏移量。