介绍

介绍

上一课大家跟福哥学会了在我们的TFLinux系统上面安装Kafka软件，今天福哥要带着大家学习使用Python去操作Kafka消息系统的方法。

Python操作Kafka可以使用pykafka这个库来实现，看这个命名是不是很眼熟？对了，类似的库名还有一个就是用来操作MySQL数据库的pymysql。pykafka库的语法和pymysql很像，不过福哥觉得Kafka和MySQL之间没有什么可比性，所以福哥选择了另外一个库。

福哥是用kafka-python这个库来操作Kafka的，这个库的设计和我们之前学习的PHP语言的rdkafka以及Java语言的KafkaConsumer/KafkaProducer对象很相似，学习起来更加舒服。

安装

直接通过pip安装kafka-python库就可以。

pip install kafka-python

使用

消费者

概念

低级消费者模式就是一个人接了一个项目，所有事情都要自己一点点做。如果当天没有做完，我会记录做到第几件了，然后第二天来了继续做下面的工作。

代码

低级消费者示例代码，Python作为控制台程序运行消费者真是太合适了。

from kafka import KafkaConsumer

topic = "test"
group_id = "lowLevel"
bootstrap_servers = "192.168.2.168:9092"
consumer = KafkaConsumer(bootstrap_servers=bootstrap_servers, group_id=group_id, auto_offset_reset="latest")
consumer.subscribe([topic])
while True:
    msg = consumer.poll(timeout_ms=12000)
    for records in msg.values():
        for i in range(0, len(records)):
            record = records[i]
            print(record.value.decode("utf-8"))

效果

生产者

代码

生产者示例代码。

from kafka import KafkaProducer
import time

topic = "test"
bootstrap_servers = "192.168.2.168:9092"
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
msg = "福哥说：现在是" + time.strftime("%H:%M:%S", time.localtime())
print(msg)
producer.send(topic, msg)

踩坑

消费者

阻塞进程

编写消费者代码的时候，福哥特别不建议使用如下两种方式，因为它会阻塞进程，按Ctrl+C都退不出来。

for方式

for msg in consumer:
    for records in msg.values():
        for i in range(0, len(records)):
            record = records[i]
            print(record.value.decode("utf-8"))

next方式

while True:
    msg = next(consumer)
    for records in msg.values():
        for i in range(0, len(records)):
            record = records[i]
            print(record.value.decode("utf-8"))

官方文档

消费者

这是官方的消费者文档

KafkaConsumer — kafka-python 2.0.2-dev documentation

生产者

这是官方的生产者文档

KafkaProducer — kafka-python 2.0.2-dev documentation

总结

今天福哥带着童鞋们学习了使用Python语言连接操作Kafka消息系统的方法，Python操作Kafka有pykafka和kafka-python两种方式，大家有空可以自行学习一下pykafka库的使用方法，然后对比一下，看看是不是和福哥有着一样的观点。