Python 操作 Kafka 增删查

Python操作Kafka可以使用kafka-python库来实现，该库提供了生产者和消费者API以及管理Kafka集群的API。

以下是Python操作Kafka的基本步骤：

1. 安装kafka-python库：可以通过pip install kafka-python来安装该库。

2. 创建生产者：使用kafka-python库中的KafkaProducer类创建一个生产者实例，然后使用send()方法向指定topic发送消息。

3. 创建消费者：使用kafka-python库中的KafkaConsumer类创建一个消费者实例，然后使用subscribe()方法订阅指定的topic并消费消息。

4. 消费消息：通过调用KafkaConsumer类的poll()方法来获取消息，然后对消息进行处理。

5. 管理Kafka集群：使用kafka-python库中的AdminClient类来管理Kafka集群，如创建topic、删除topic等。

下面是Python操作Kafka的示例代码：

from kafka import KafkaProducer, KafkaConsumer, AdminClient, TopicPartition

# 创建生产者
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('test-topic', b'Hello, Kafka!')

# 创建消费者
consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'], group_id='test-group')
for message in consumer:
    print(message)

# 管理Kafka集群
admin_client = AdminClient(bootstrap_servers=['localhost:9092'])
admin_client.create_topics(new_topics=[TopicPartition(topic='test-topic', partition=0)])
admin_client.delete_topics(topics=['test-topic'])