关于近期学习kafka总结

从最早以前开始消费，不需要传递group_id参数，因为group_id是用于记录偏移量的。只需要设置

auto_offset_reset="earliest"

from kafka import KafkaConsumer

kafka_addr = '192.168.19.131:9092'

def func():
    consumer = KafkaConsumer('quickstart',bootstrap_servers=kafka_addr,auto_offset_reset="earliest",
                        )
    print(consumer.topics())
                            
    for res in consumer:
        print(res)

while True:
    func()

从最近的未消费消息开始消费需要指定group_id,因为group_id记录了偏移量offset，不需要传递auto_offset_reset参数，因为默认就是auto_offset_reset=latest

from kafka import KafkaConsumer

kafka_addr = '192.168.19.131:9092'

def func():
    consumer = KafkaConsumer('quickstart',group_id='test',bootstrap_servers=kafka_addr,
                        )

    print(consumer.topics())

    for res in consumer:
        print(res)

while True:
    func()