python - 如何在程序中停止 Python Kafka Consumer？

Question

我正在做 Python Kafka 消费者(尝试在 http://kafka-python.readthedocs.org/en/latest/apidoc/kafka.consumer.html 中使用 kafka.consumer.SimpleConsumer 或 kafka.consumer.simple.SimpleConsumer)。当我运行下面这段代码时，它会一直运行，即使所有的消息都被消耗掉了。 希望消费者消费完所有的消息就停下来。怎么做？我也不知道如何使用 stop() 函数(在基类 kafka.consumer.base.Consumer 中)。

更新

我使用信号处理程序来调用 consumer.stop()。一些错误消息被打印到屏幕上。但是程序仍然卡在for循环中。当新消息进来时，消费者消费它们并打印它们。我也试过 client.close()。但结果相同。

我需要一些方法来优雅地停止 for 循环。

        client = KafkaClient("localhost:9092")
        consumer = SimpleConsumer(client, "test-group", "test")

        consumer.seek(0, 2)# (0,2) and (0,0)

        for message in consumer:
            print "Offset:", message.offset
            print "Value:", message.message.value

欢迎任何帮助。谢谢。

Answer 1

我们可以先查看主题中最后一条消息的偏移量。 然后在我们达到该偏移量时停止循环。

    client = "localhost:9092"
    consumer = KafkaConsumer(client)
    topic = 'test'
    tp = TopicPartition(topic,0)
    #register to the topic
    consumer.assign([tp])

    # obtain the last offset value
    consumer.seek_to_end(tp)
    lastOffset = consumer.position(tp)

    consumer.seek_to_beginning(tp)        

    for message in consumer:
        print "Offset:", message.offset
        print "Value:", message.message.value
        if message.offset == lastOffset - 1:
            break