使用python编写程序实现kafka消费并保证数据一致性

以下是使用Python编写程序实现Kafka消费并保证数据一致性的示例代码：

from kafka import KafkaConsumer
from kafka import TopicPartition

consumer = KafkaConsumer(bootstrap_servers=['localhost:9092'],
                         group_id='my-group',
                         auto_offset_reset='earliest')

# 订阅主题和分区
tp = TopicPartition('my-topic', 0)
consumer.assign([tp])

# 获取当前分区的最新偏移量
end_offset = consumer.end_offsets([tp])[tp]

# 获取上一次消费的偏移量
with open('offset.txt', 'r') as f:
    last_offset = int(f.read())

# 从上一次消费的偏移量开始消费
consumer.seek(tp, last_offset)

# 处理消息
for message in consumer:
    # 处理消息
    print(message)

    # 保存偏移量
    with open('offset.txt', 'w') as f:
        f.write(str(message.offset + 1))

    # 确认消息已经消费完成
    consumer.commit()

在上面的代码中，我们使用KafkaConsumer创建一个消费者实例，并订阅主题和分区。我们使用TopicPartition指定我们要消费的分区，并从上一次消费的偏移量开始消费。在处理完每个消息后，我们将消费的偏移量保存到文件中，并使用commit确认消息已经消费完成。

这样，即使我们的消费者崩溃或重启，它也能够从上一次消费的偏移量开始恢复，并保证数据一致性。