如何用Python操作Kafka？

pip install confluent-kafka

from confluent_kafka import Producer

# 配置Kafka生产者的参数
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'python-producer'
}

# 创建生产者
producer = Producer(conf)

# 生产消息到topic
def delivery_report(err, msg):
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

topic = 'my_topic'
for i in range(10):
    producer.produce(topic, key=str(i), value=f'Message {i}')
    producer.poll(0)

producer.flush()

from confluent_kafka import Consumer, KafkaException

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'python-consumer',
    'auto.offset.reset': 'earliest'
}

# 创建消费者
consumer = Consumer(conf)

# 订阅topic
consumer.subscribe(['my_topic'])

# 消费消息
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # End of partition event
                print('%% %s [%d] reached end at offset %d\n' %
                      (msg.topic(), msg.partition(), msg.offset()))
            elif msg.error():
                raise KafkaException(msg.error())
        else:
            print('Received message: {}'.format(msg.value().decode('utf-8')))

except KeyboardInterrupt:
    pass
finally:
    # 关闭消费者
    consumer.close()