使用Python的SimpleConsumer()实现Kafka消息消费的断点续传

使用Python的SimpleConsumer()实现Kafka消息消费的断点续传功能是通过记录消费的消息的偏移量来实现的。简单来说，当一个消费者消费到一个消息时，它会记录这个消息的偏移量，在下一次消费时，会从记录的偏移量开始消费。

以下是一个使用Python的SimpleConsumer()实现Kafka消息消费的断点续传的示例代码：

from kafka import KafkaConsumer
import json

# Kafka服务地址
bootstrap_servers = ['localhost:9092']
# 消费者组ID
group_id = 'my-group'
# 订阅的topic名称
topic_name = 'my-topic'
# 断点记录文件路径
offset_file_path = 'offset.json'

# 创建一个Kafka消费者
consumer = KafkaConsumer(topic_name, group_id=group_id, bootstrap_servers=bootstrap_servers)

# 读取断点记录文件，获取上次消费的偏移量
try:
    with open(offset_file_path, 'r') as offset_file:
        offset_data = json.load(offset_file)
        consumer.seek(offset_data['topic_parti'], offset_data['offset'])
except FileNotFoundError:
    pass

# 消费消息
for message in consumer:
    # 处理消息逻辑
    print("Received message: ", message.value.decode('utf-8'))
    
    # 记录本次消费的偏移量
    offset_data = {
        'topic_parti': message.partition,
        'offset': message.offset
    }
    with open(offset_file_path, 'w') as offset_file:
        json.dump(offset_data, offset_file)

在上述示例代码中，首先根据配置创建了一个Kafka消费者对象，然后读取了断点记录文件以获取上次消费的偏移量。然后通过调用seek()方法将消费者的偏移量设置为上次记录的偏移量，这样消费者就会从上次消费的位置开始消费。

接下来使用一个for循环不断地从Kafka消费者接收消息，并处理消息逻辑。在处理完每条消息后，将本次消费的偏移量记录到断点记录文件中，以便下次断点续传时使用。

需要注意的是，如果 次运行或没有断点记录文件时，代码会抛出FileNotFoundError异常，可以通过捕获异常的方式来处理，例如在 次运行时创建一个空的断点记录文件。

以上就是使用Python的SimpleConsumer()实现Kafka消息消费的断点续传的示例代码。通过记录偏移量可以实现消息消费的断点续传，即使消费者停止后再次启动也能从上次停止的位置继续消费。