获取Python中汇合的Kafka主题的最新消息

from confluent_kafka import Consumer, KafkaException

def create_consumer(broker, group_id, topic):
    conf = {
        'bootstrap.servers': broker,
        'group.id': group_id,
        'auto.offset.reset': 'latest'  # 从最新的消息开始消费
    }
    consumer = Consumer(conf)
    consumer.subscribe([topic])
    return consumer

def consume_messages(consumer, timeout=1.0):
    try:
        while True:
            msg = consumer.poll(timeout)
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())
            else:
                print(f"Received message: {msg.value().decode('utf-8')}")
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()

if __name__ == "__main__":
    broker = 'localhost:9092'  # Kafka 代理地址
    group_id = 'my-group'      # 消费者组ID
    topic = 'my-topic'         # 主题名称

    consumer = create_consumer(broker, group_id, topic)
    consume_messages(consumer)