面试题：Python实现Redis数据增量备份与恢复

设计思路

标记已备份数据：
- 可以在Redis中维护一个特殊的键值对，记录上次备份的时间戳。每次备份完成后，更新这个时间戳。例如，使用键backup:last_timestamp来存储时间戳。
检测数据变化：
- 利用Redis的CLIENT LIST命令获取客户端连接信息，结合MONITOR命令可以监控Redis服务器接收到的所有写命令。对于每一个写命令，记录其执行时间。当进行备份时，对比命令执行时间和上次备份的时间戳，大于上次备份时间戳的命令所涉及的数据就是变化的数据。
- 另一种方法是使用Redis的SCAN命令遍历所有键，并结合OBJECT命令获取每个键的最后访问时间（在某些Redis版本中支持），通过对比最后访问时间和上次备份时间戳来确定数据是否变化。
备份功能：
- 当检测到变化的数据后，将这些数据备份到文件系统或者其他持久化存储中。可以将数据以JSON等格式存储，方便后续恢复。
恢复功能：
- 读取备份文件，解析其中的数据，然后使用Redis的相关命令将数据重新写入Redis。

关键代码片段

获取上次备份时间戳：

import redis

r = redis.Redis(host='localhost', port=6379, db = 0)

def get_last_backup_timestamp():
    timestamp = r.get('backup:last_timestamp')
    if timestamp is None:
        return 0
    return float(timestamp)

检测数据变化并备份（简单示例，假设通过对比键的最后访问时间）：

import time

def backup_changed_data():
    last_timestamp = get_last_backup_timestamp()
    all_keys = []
    cursor = '0'
    while cursor != 0:
        cursor, keys = r.scan(cursor = cursor)
        all_keys.extend(keys)

    changed_data = {}
    for key in all_keys:
        object_info = r.object('idletime', key)
        access_time = time.time() - object_info / 1000  # 转换为秒
        if access_time > last_timestamp:
            value = r.get(key)
            changed_data[key.decode('utf - 8')] = value.decode('utf - 8') if isinstance(value, bytes) else value

    # 备份数据到文件
    with open('backup.json', 'w') as f:
        import json
        json.dump(changed_data, f)

    # 更新备份时间戳
    r.set('backup:last_timestamp', time.time())

恢复数据：

def restore_data():
    try:
        with open('backup.json', 'r') as f:
            import json
            data = json.load(f)
        for key, value in data.items():
            r.set(key, value)
    except FileNotFoundError:
        print("Backup file not found.")

星途面试题库

面试题：Python实现Redis数据增量备份与恢复

知识考点

面试题答案

设计思路

关键代码片段