python实现kafaka消息的发送和消费代码示例

我就是我 2024-02-20 13:12 51阅读 0赞

以下是使用 Python 进行 Kafka 消息发送和消费的示例代码：

首先，确保已经安装了 `kafka-python` 库，可以使用以下命令进行安装：

pip install kafka-python

1.  Kafka 消息发送示例：

from kafka import KafkaProducer
    
    # 创建 KafkaProducer 实例
    producer = KafkaProducer(bootstrap_servers='localhost:9092')
    
    # 发送消息到指定 Topic
    topic = 'my_topic'
    message = 'Hello, Kafka!'
    
    # 发送消息
    producer.send(topic, message.encode())
    producer.flush()  # 等待消息发送完成
    print("消息发送成功")
    
    # 关闭 KafkaProducer
    producer.close()

在上述代码中，我们首先创建了一个 KafkaProducer 实例，指定了 Kafka 服务器的地址（bootstrap\_servers）。然后，我们定义了要发送的消息内容和要发送到的 Topic。使用 `producer.send()` 方法将消息发送到指定的 Topic，通过调用 `producer.flush()` 来确保消息发送完成，并最后关闭 KafkaProducer。

1.  Kafka 消息消费示例：

from kafka import KafkaConsumer
    
    # 创建 KafkaConsumer 实例
    consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092')
    
    # 消费消息
    try:
        while True:
            messages = consumer.poll(timeout_ms=500)  # 每500毫秒拉取一次消息
            for topic_partition, message_list in messages.items():
                for message in message_list:
                    print(f"收到消息: {
          message.value.decode()}")
    
    except KeyboardInterrupt:
        pass
    
    # 关闭 KafkaConsumer
    consumer.close()

在上述代码中，我们创建了一个 KafkaConsumer 实例，并指定要消费的 Topic。通过迭代 KafkaConsumer 对象，我们可以遍历消费者接收到的消息并进行处理。在这个示例中，我们简单地将收到的消息打印出来。

请注意，以上示例中的 Kafka 服务器地址为 `'localhost:9092'`，你可以根据实际情况修改为正确的地址。另外，确保主题（Topic）的名称和消费者订阅的主题名称一致。