Python连接Kafka收发数据等操作

一、Kafka

Apache Kafka 是一个开源流处理平台，由 LinkedIn 开发，并于 2011 年成为 Apache 软件基金会的一部分。Kafka 广泛用于构建实时的数据流和流式处理应用程序，它以高吞吐量、可扩展性和容错性著称。

kafka-python 是一个用 Python 编写的 Apache Kafka 客户端库。

安装命令如下：

pip install kafka-python

二、发送端（生产者）

自动创建test主题，并每隔一秒发送一条数据，示例代码如下：

from kafka import KafkaProducer
import json
import time

# Kafka服务器地址
bootstrap_servers = ['localhost:9092']

# 创建KafkaProducer实例
producer = KafkaProducer(bootstrap_servers=bootstrap_servers)

# 发送消息的函数
def send_message(topic, message):
    # 将消息转换为字节
    producer.send(topic, json.dumps(message).encode('utf-8'))
    producer.flush()

if __name__ == '__main__':
    # 创建'test'主题
    topic = 'test'
    # 发送消息
    i = 1
    while True:
        message = {'num': i, 'msg': f'Hello Kafka {i}'}
        send_message(topic, message)
        i += 1
        time.sleep(1)

三、接收端（消费者）

代码如下：

from kafka import KafkaConsumer
import json

# Kafka服务器地址
bootstrap_servers = ['localhost:9092']

# 创建KafkaConsumer实例
consumer = KafkaConsumer(
    'test',
    bootstrap_servers=bootstrap_servers,
    auto_offset_reset='latest',  # 从最新的消息开始消费
    # auto_offset_reset='earliest',  # 从最早的offset开始消费
    enable_auto_commit=True,  # 自动提交offset
    group_id='my-group'  # 消费者组ID
)

# 消费消息
for message in consumer:
    # 将接收到的消息解码并转换为字典
    message = json.loads(message.value.decode('utf-8'))
    print(f"Received message: {message}")

消费者参数如下：

1、auto_offset_reset
该参数指定了当Kafka中没有初始偏移量或当前偏移量在服务器上不再存在时（例如数据被删除了），消费者应从何处开始读取数据。
可选值：
earliest：从最早的记录开始消费，即从分区日志的开始处开始。
latest：从最新的记录开始消费，即从分区日志的末尾开始。（默认）
none：如果没有为消费者指定初始偏移量，就抛出一个异常。

2、enable_auto_commit

该参数指定了消费者是否周期性地提交它所消费的偏移量。自动提交偏移量可以简化消费者的使用，但可能有重复消费或数据丢失的风险。禁用自动提交可以更精确地控制偏移量的提交时机，通常在确保消息处理成功后才提交偏移量。
可选值：
true：自动提交偏移量。（默认）
false：不自动提交偏移量，需要手动调用commitSync()或commitAsync()来提交偏移量。

3、group_id

该参数用于指定消费者所属的消费组。同一个消费组的消费者将共同消费一个主题的不同分区，而不同消费组的消费者可以独立地消费消息，互不影响。这对于实现负载均衡和故障转移很有用。
类型：字符串（必须指定）

四、其他操作

list_topics()：获取主题元数据。

create_topics()：创建新主题。

delete_topics()：删除主题。

from kafka.admin import KafkaAdminClient, NewTopic

# 获取主题元数据
admin_client = KafkaAdminClient(bootstrap_servers='localhost:9092', client_id='test')
topics = admin_client.list_topics()
print(topics)

# 创建主题
new_topic = NewTopic(name="test-topic", num_partitions=3, replication_factor=1)
admin_client.create_topics(new_topics=[new_topic], validate_only=False)

# 删除主题
admin_client.delete_topics(topics=['test-topic'])