Python使用pykafka测试Kafka集群的完整示例

程一诺

发布时间：2026-10-13| 5人浏览过

原创

本文介绍了如何使用Python的pykafka库连接Kafka集群。通过提供生产者和消费者的完整代码示例，展示了如何配置主机地址、创建客户端、发送消息以及消费消息。该实例适用于希望快速验证Kafka集群连通性及基础功能的开发者，涵盖了从环境配置到核心API调用的关键步骤。

Kafka生产者代码实现

以下是用于向Kafka集群发送消息的生产者代码实现。


# -* coding:utf8 *- 
from pykafka import KafkaClient 
 
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host) 
 
print client.topics 
 
# 生产者 
topicdocu = client.topics['my-topic'] 
producer = topicdocu.get_producer() 
for i in range(100): 
  print i 
  producer.produce('test message ' + str(i ** 2)) 
producer.stop()

Kafka消费者代码实现

以下是用于从Kafka集群接收消息的消费者代码实现。


# -* coding:utf8 *- 
from pykafka import KafkaClient 
 
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host) 
 
print client.topics 
 
# 消费者 
topic = client.topics['my-topic'] 
consumer = topic.get_simple_consumer(consumer_group='test', auto_commit_enable=True, auto_commit_interval_ms=1, 
                   consumer_id='test') 
for message in consumer: 
  if message is not None: 
    print message.offset, message.value

上述内容提供了Python测试Kafka集群（基于pykafka库）的完整示例。希望这些代码能为您提供参考，助您顺利完成相关开发工作。如果您觉得有用，欢迎支持我们的网站。