Python 使用kafka

1. 简介

Kafka 是一个分布式流处理平台，由 LinkedIn 公司开发并开源。它主要用于解决大规模数据的实时消费与处理问题。Kafka 具有高吞吐量、可持久化、分布式、可伸缩、高可靠等特点，已经成为了处理实时数据流的重要工具之一。

Python 作为一门流行的编程语言，提供了丰富的库和工具来支持与 Kafka 之间的交互。本文将介绍如何使用 Python 连接和操作 Kafka 消息队列。

2. 安装 Kafka-python

在使用 Python 连接 Kafka 之前，需要安装 kafka-python 库。可以使用 pip 命令来安装：

pip install kafka-python

3. 连接到 Kafka

使用 kafka-python 库连接到 Kafka 集群非常简单。下面是一个示例代码：

from kafka import KafkaConsumer, KafkaProducer

# 创建 Kafka Producer
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda x: str(x).encode('utf-8')
)

# 发送消息
producer.send('my_topic', value='Hello Kafka!')

# 创建 Kafka Consumer
consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='my_group',
    value_deserializer=lambda x: x.decode('utf-8')
)

# 消费消息
for message in consumer:
    print(message.value)

在上面的代码中，我们使用 KafkaProducer 创建一个 Kafka 生产者并发送消息到名为 my_topic 的主题。然后使用 KafkaConsumer 创建一个 Kafka 消费者并消费 my_topic 主题中的消息。

4. 发送消息

使用 kafka-python 发送消息也非常简单。可以使用 send 方法来发送消息到指定的主题。下面是一个示例代码：

from kafka import KafkaProducer

# 创建 Kafka Producer
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda x: str(x).encode('utf-8')
)

# 发送消息
producer.send('my_topic', value='Hello Kafka!', key='1')

# 发送多条消息
producer.send('my_topic', value='Message 1', key='1')
producer.send('my_topic', value='Message 2', key='2')
producer.send('my_topic', value='Message 3', key='3')

# 异步发送消息
future = producer.send('my_topic', value='Async Message')
result = future.get(timeout=10)
print(result)

上述代码中，我们创建了一个 Kafka Producer，并使用 send 方法发送了多条消息到主题 my_topic 中。可以指定消息的 key，方便对消息进行分区。也可以使用异步方式发送消息，在此示例中，我们使用 future.get(timeout=10) 等待消息发送结果并打印。

5. 消费消息

使用 kafka-python 消费消息也非常简单。可以使用 KafkaConsumer 创建一个消费者并使用 for 循环来遍历消息。下面是一个示例代码：

from kafka import KafkaConsumer

# 创建 Kafka Consumer
consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='my_group',
    value_deserializer=lambda x: x.decode('utf-8')
)

# 消费消息
for message in consumer:
    print(message.value)

上述代码中，我们创建了一个 Kafka Consumer，并使用 for 循环遍历消费 my_topic 主题中的消息。可以根据需要设置消费者的配置，例如 auto_offset_reset 用于设置消费者的起始消费位置，enable_auto_commit 用于设置是否自动提交消费偏移量等。

6. 手动提交偏移量

在 Kafka 中，消费者需要手动提交消费偏移量，以确保消息被正确消费。以下是一个示例代码，展示了如何手动提交偏移量：

from kafka import KafkaConsumer

# 创建 Kafka Consumer
consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers='localhost:9092',
    enable_auto_commit=False,
    value_deserializer=lambda x: x.decode('utf-8')
)

# 消费消息
for message in consumer:
    print(message.value)
    consumer.commit()

在上述代码中，我们设置 enable_auto_commit=False 来禁止自动提交消费偏移量。然后在消费完消息后，调用 commit 方法手动提交偏移量。

7. 使用 Avro Schema

Kafka 支持使用 Avro Schema 来序列化和反序列化消息。Avro 是一种二进制的数据序列化格式，提供了数据结构的版本管理、动态解析等特性。要在 Python 中使用 Avro，需要安装 avro-python3 库。

下面是一个使用 Avro Schema 的示例代码：

from kafka import KafkaProducer, KafkaConsumer
from avro import schema, io
import avro.schema
import avro.io

# 定义 Avro Schema
avro_schema = avro.schema.Parse(open("user.avsc", "rb").read())

# 创建 Kafka Producer
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda x: avro.io.DatumWriter(avro_schema).write(x)
)

# 发送消息
message = {"name": "Alice", "age": 25}
producer.send('my_topic', value=message)

# 创建 Kafka Consumer
consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='my_group',
    value_deserializer=lambda x: avro.io.DatumReader(avro_schema).read(x)
)

# 消费消息
for message in consumer:
    print(message.value)