Redis模糊匹配删除Key：从SCAN原理到生产级Python脚本实现

1. 项目概述为什么需要模糊匹配删除Key在Redis的日常运维和开发中我们经常会遇到一种场景需要批量清理一批具有特定命名模式的Key。比如某个微服务下线了遗留了大量以user:session:为前缀的会话缓存或者一个临时的活动结束了所有以campaign:2024Q3:开头的活动数据需要被清除。手动一个个去DEL显然不现实而Redis本身并没有提供一个直接的DEL pattern命令。这时“模糊匹配并删除Key”就成了一个高频且刚需的操作。这个需求背后反映的是缓存治理和数据生命周期管理的核心问题。无效或过期的Key不仅占用宝贵的内存资源在集群模式下还可能影响Slot的均衡分布甚至在某些极端情况下大量无用的Key遍历操作会拖慢KEYS命令尽管生产环境严禁使用影响服务性能。因此掌握一套安全、高效、可脚本化的模糊删除方法是每一位Redis使用者必备的技能。它不仅仅是执行一条命令更涉及到对Redis单线程模型、阻塞风险、生产环境禁忌以及替代方案的深刻理解。2. 核心思路与方案选型从KEYS到SCAN当提到“模糊匹配”很多人的第一反应是KEYS pattern命令。这很自然因为它最直观。KEYS user:session:*一下就能列出所有匹配的Key。然而在任何一个稍有规模的线上Redis服务中直接使用KEYS命令都是被明令禁止的。原因在于KEYS命令会一次性遍历整个数据库的所有Key在Key数量巨大时这个操作会阻塞Redis的单线程导致在此期间所有其他命令都无法执行服务呈现“假死”状态对线上业务是灾难性的。因此我们必须寻找非阻塞的、增量式的迭代方案。这就是SCAN命令家族存在的意义。SCAN及其针对特定数据类型的变种如SSCAN,HSCAN,ZSCAN提供了一种游标式的迭代方式每次只返回一小部分元素不会长时间阻塞服务器。我们的核心思路也随之清晰使用SCAN命令迭代匹配出所有目标Key然后逐个或批量执行DEL。基于这个核心思路衍生出几种常见的实现方案Redis CLI 管道Pipe操作直接在服务器上使用redis-cli结合SCAN和xargs进行删除。适合运维人员临时手动操作。Lua 脚本将SCAN和DEL的逻辑写入一个Lua脚本在服务器端原子性执行。优点是减少网络往返避免部分删除的中间状态但脚本执行过长同样会阻塞。客户端程序用Python、Java、Go等语言编写小程序利用客户端的SCANAPI进行迭代和批量删除。灵活性最高可以加入复杂的逻辑控制、日志和错误处理是我们重点讨论的方案。对于生产环境我强烈推荐使用客户端程序的方式。它兼具了安全可控的迭代、高效可批量删除和灵活可记录、可重试的优点。下文将主要以Python为例拆解其中的技术细节和避坑指南。3. 工具与依赖准备“工欲善其事必先利其器。” 在开始编写删除脚本之前我们需要准备好相应的环境和工具。这里假设你已经在目标服务器上安装了Redis并且可以通过网络进行连接。3.1 客户端库选择以Python为例最常用的Redis客户端是redis-py。你可以通过pip轻松安装pip install redis对于生产环境建议锁定一个稳定的版本例如redis4.0.0, 5.0.0以避免API变动带来的意外问题。redis-py4.x版本提供了良好的异步支持和连接池管理。3.2 连接配置与安全考量在脚本中创建Redis连接时切忌将密码等敏感信息硬编码。推荐使用配置文件或环境变量来管理连接参数。import os import redis # 从环境变量读取配置更安全 REDIS_HOST os.getenv(REDIS_HOST, localhost) REDIS_PORT int(os.getenv(REDIS_PORT, 6379)) REDIS_PASSWORD os.getenv(REDIS_PASSWORD, None) REDIS_DB int(os.getenv(REDIS_DB, 0)) # 创建连接池提升性能 pool redis.ConnectionPool( hostREDIS_HOST, portREDIS_PORT, passwordREDIS_PASSWORD, dbREDIS_DB, decode_responsesTrue, # 自动将返回的bytes解码为str处理起来更方便 socket_connect_timeout5, socket_timeout10 ) r redis.Redis(connection_poolpool) # 测试连接 try: r.ping() print(Redis连接成功) except redis.exceptions.ConnectionError as e: print(fRedis连接失败: {e}) exit(1)注意decode_responsesTrue这个参数非常实用它让客户端自动将Redis返回的二进制数据转为Python字符串省去了我们手动.decode(utf-8)的麻烦。但如果你存储的值本身就是二进制数据如图片则不要开启此选项。3.3 权限与风险确认在执行批量删除这种“危险”操作前务必进行二次确认确认当前客户端账号权限执行删除操作的账号是否只有目标数据库的权限最好不要使用最高权限的账号。确认目标数据库SELECT命令切换数据库的用法在集群模式下是不支持的。你的连接是否指向了正确的DB可以通过r.client_list()查看当前连接信息或者直接r.get(一个测试key)来验证。备份意识如果待删除的Key价值很高是否有备份或快照虽然Redis的删除不可逆但如果你有最新的RDB或AOF备份在误操作后还有挽回的余地尽管恢复起来很麻烦。4. 核心实现SCAN迭代与批量删除这是整个脚本最核心的部分。我们的目标是安全、完整、高效地删除所有匹配模式的Key。4.1 SCAN命令深度解析首先我们要彻底理解SCAN命令。它的基本语法是SCAN cursor [MATCH pattern] [COUNT count]。cursor游标第一次调用时传入0之后传入上一次调用返回的游标。MATCH pattern匹配模式支持*(任意字符),?(单个字符),[abc](字符集) 等通配符。COUNT count一个提示性的参数并非精确返回数量。它只是告诉Redis“我希望你每次大概扫描这么多元素”。实际返回的数量可能多于、少于或等于count。默认值是10。增大COUNT可以提高迭代速度但单次命令的响应时间会变长需要权衡。一个关键特性是SCAN在整个迭代过程中如果数据集没有发生大的变化如扩容它可以保证返回所有在迭代开始时就存在的元素但可能会返回重复的元素。客户端需要自己处理去重。4.2 基础迭代删除函数下面是一个稳健的基础实现函数def delete_keys_by_pattern(pattern, batch_size100): 根据模式模糊匹配并删除Key :param pattern: 匹配模式如 user:session:* :param batch_size: 每次SCAN命令的COUNT提示值 :return: 删除的Key总数 deleted_count 0 # 用于记录已处理的游标防止逻辑错误 cursor 0 # 可以使用一个集合来存储已发现的Key用于最终核对或防止极端情况下的重复删除 # all_keys set() while cursor ! 0: # 在Python redis-py中scan返回一个元组 (next_cursor, key_list) cursor, keys r.scan(cursorcursor, matchpattern, countbatch_size) if keys: # 批量删除使用pipeline减少网络往返 pipe r.pipeline(transactionFalse) # 对于批量DEL不需要事务 for key in keys: pipe.delete(key) results pipe.execute() # 结果是一个列表每个元素是对应delete操作删除的数量1或0 deleted_in_batch sum(results) # 计算这一批实际删除了多少个 deleted_count deleted_in_batch print(f批次删除 {len(keys)} 个Key实际成功删除 {deleted_in_batch} 个。当前累计: {deleted_count}) else: print(f游标 {cursor} 未匹配到Key。) # 注意当SCAN返回的下一个游标为0时表示迭代结束 # 但redis-py将返回的字节游标0转换成了整数0所以我们这里用整数0判断 # 更严谨的写法是if cursor 0: break print(f删除完成。总计删除 {deleted_count} 个Key。) return deleted_count4.3 关键参数调优与陷阱COUNT参数的选择这不是“每次返回多少Key”而是“每次扫描多少桶slot”。Redis内部哈希表的结构决定了这一点。对于百万级Key的数据库COUNT设为1000可能比较合适。你可以写一个测试脚本固定匹配一个不存在的模式如__test_nonexistent__*调整COUNT值观察单次SCAN的耗时和网络包大小找到一个适合你当前网络和Redis负载的平衡点。一个经验值是500-2000。游标归零的判断SCAN命令返回的游标是一个字符串。当返回0时表示迭代结束。在redis-py中如果设置了decode_responsesTrue游标会被转换成字符串0如果未设置则是字节b0。我们的循环条件while cursor ! 0能工作是因为redis-py在内部做了转换当游标为0时下一次调用scan(cursor0)会返回(0, [])从而退出循环。但最清晰的写法是cursor 0 while True: cursor, keys r.scan(cursorcursor, matchpattern, countbatch_size) # ... 处理keys ... if cursor 0: break“幻影Key”问题在漫长的SCAN迭代过程中可能会有新的Key被创建并匹配模式也可能有Key被删除。SCAN不保证能返回这些在迭代开始后变动的Key。这是最终一致性而非强一致性。对于删除操作这通常可以接受因为我们的目标是清理“当前”的无效数据。如果要求绝对精确可能需要短暂停止写入或者迭代多次直到连续两次结果为空。5. 生产级脚本增强稳健性与可观测性上面的基础函数可以工作但对于生产环境还远远不够。我们需要给它穿上“盔甲”。5.1 添加延迟与熔断机制长时间、高频率的SCAN和DEL操作本身也会对Redis造成一定压力。为了不影响线上正常业务我们需要给脚本加上“温柔”的节奏。import time def delete_keys_safely(pattern, batch_size100, delay0.01, max_keys_per_runNone): 安全删除Key支持延迟控制和总量限制 :param delay: 每批操作后的休眠时间秒用于降低对Redis的瞬时压力 :param max_keys_per_run: 本次运行最多删除的Key数量用于分时段执行 deleted_count 0 cursor 0 total_scanned 0 try: while True: cursor, keys r.scan(cursorcursor, matchpattern, countbatch_size) total_scanned len(keys) if keys: # 1. 检查本次运行是否已达到上限 if max_keys_per_run and deleted_count len(keys) max_keys_per_run: # 只删除上限内的部分 keys_to_delete keys[:max_keys_per_run - deleted_count] print(f达到单次运行上限 {max_keys_per_run}本批次仅处理 {len(keys_to_delete)} 个Key。) else: keys_to_delete keys # 2. 执行删除 pipe r.pipeline() for key in keys_to_delete: pipe.delete(key) results pipe.execute() batch_deleted sum(results) deleted_count batch_deleted print(f扫描到 {len(keys)} 个本批处理 {len(keys_to_delete)} 个成功删除 {batch_deleted} 个。累计删除: {deleted_count} 总扫描: {total_scanned}) # 3. 检查是否达到运行上限 if max_keys_per_run and deleted_count max_keys_per_run: print(f已达到单次运行上限 {max_keys_per_run}停止执行。) break # 4. 延迟减轻服务器压力 if delay 0: time.sleep(delay) if cursor 0: print(迭代结束。) break except redis.exceptions.RedisError as e: print(fRedis操作发生错误: {e}) # 这里可以记录错误甚至将当前游标保存到文件以便后续重试 save_cursor_to_file(cursor) except KeyboardInterrupt: print(\n用户中断操作。) print(f已删除 {deleted_count} 个Key最后游标: {cursor}) # 同样可以保存游标 return deleted_count, total_scanneddelay参数即使设为0.0011毫秒对于每秒数万QPS的Redis来说也能显著平滑流量曲线避免请求突增。5.2 日志、监控与进度保存一个健壮的脚本必须可观测、可恢复。日志不要只打印到控制台。应使用Python的logging模块将运行日志开始时间、模式、每批结果、错误信息、结束时间输出到文件。这对于事后审计和问题排查至关重要。进度保存如果脚本因为网络中断、人为终止等原因停止我们希望能从断点继续。可以将当前的游标cursor和已删除数量deleted_count定期比如每处理1000个Key保存到本地文件或一个特殊的Redis Key里。重启脚本时先读取这个进度然后从保存的游标开始继续SCAN。监控在脚本运行期间可以定期如每删除5000个Key检查Redis的info memory和info stats观察内存下降情况以及evicted_keys如果配置了最大内存等指标确认删除操作正在生效且没有引发异常。5.3 模拟执行Dry Run与确认机制这是生产环境操作前的绝对必要步骤在真正执行删除前必须先进行一次“模拟演练”只查找Key不删除。def dry_run_scan(pattern, batch_size100, max_samples1000): 模拟执行只扫描并列出Key不删除。用于确认匹配范围。 :param max_samples: 最大采样数量避免输出过多 found_keys [] cursor 0 sample_count 0 print(f开始模拟扫描模式: {pattern}) try: while True and sample_count max_samples: cursor, keys r.scan(cursorcursor, matchpattern, countbatch_size) for key in keys: if sample_count max_samples: # 可以尝试获取Key的类型和TTL获取更详细信息 key_type r.type(key) ttl r.ttl(key) found_keys.append((key, key_type, ttl)) sample_count 1 else: print(f已达到最大采样数 {max_samples}停止扫描。) break if cursor 0: break except Exception as e: print(f扫描过程中发生错误: {e}) print(f\n模拟扫描完成。共发现约 {len(found_keys)} 个Key采样前{max_samples}个:) for i, (key, key_type, ttl) in enumerate(found_keys[:20]): # 只打印前20个作为示例 ttl_info f, TTL: {ttl}s if ttl 0 else , 永不过期 if ttl -1 else , 已过期 print(f {i1}. {key} (类型: {key_type}{ttl_info})) if len(found_keys) 20: print(f ... 以及另外 {len(found_keys)-20} 个Key。) # 询问确认 confirm input(f\n确认要删除所有匹配 {pattern} 的Key吗(请输入 YES 以继续): ) return confirm.strip().upper() YES在执行真正的删除函数前先调用dry_run_scan仔细检查打印出的Key样例确保它们都是你“想删”的而不是“不能删”的。确认无误后再输入“YES”执行删除。6. 高级场景与特殊处理6.1 处理大数据KeyList, Hash, Set, Sorted Set如果要删除的Key对应的是很大的List、Hash等直接DEL是没问题的因为Redis会负责回收内存。但如果你在删除前还需要处理其中的数据比如备份就需要使用SSCAN、HSCAN、ZSCAN来遍历大Key内部的元素。不过对于单纯的删除操作DEL命令足够了。6.2 集群模式下的挑战在Redis Cluster中Key分布在不同的Slot上而SCAN命令每次只针对当前连接的节点。要在集群中全局模糊删除有两种主流方法遍历所有主节点通过CLUSTER SLOTS或CLUSTER NODES命令获取集群所有主节点的地址然后依次连接到每个节点在该节点上执行上述的SCAN和DEL逻辑。这是最彻底的方法。# 获取集群节点信息假设已有一个集群客户端连接 cluster_client nodes cluster_client.cluster_nodes() masters [node for node in nodes if master in node[flags]] for master in masters: host, port master[host], master[port] node_client redis.Redis(hosthost, portport, passwordREDIS_PASSWORD) # 在这个节点客户端上调用 delete_keys_by_pattern delete_keys_on_node(node_client, pattern)使用redis-cli的--cluster call命令这是Redis官方推荐的方式非常方便。redis-cli -c -h cluster-host -p cluster-port --cluster call node-host:node-port SCAN 0 MATCH user:session:* COUNT 1000 | grep -oP (?\[)[^\]]*(?\]) | xargs -L 1 redis-cli -c -h cluster-host -p cluster-port del这条命令的原理是让集群所有节点执行SCAN收集Key然后统一删除。但命令构造复杂容易出错且错误处理能力弱。对于生产环境强烈建议用第一种方法编写可控的脚本。6.3 与过期策略结合有时我们想删除的其实是那些“已经过期”但Redis尚未主动清理的Key。Redis的过期删除是惰性和定期两种策略结合可能存在已过期的Key仍占用内存的情况。我们可以结合SCAN和TTL命令if keys: pipe r.pipeline() for key in keys: # 只删除TTL为-2已过期或-1永不过期中符合特定条件的Key # 或者删除TTL大于某个阈值的Key ttl r.ttl(key) # 注意这里每个Key一次网络往返性能差 if ttl -2: # -2 表示Key不存在可能刚被删除-1表示永不过期0表示剩余生存时间 pipe.delete(key) pipe.execute()但请注意为每个Key单独查询TTL会产生大量网络请求性能极差。更好的方法是如果业务允许直接给这些Key设置一个统一的过期时间EXPIRE让Redis自己去清理而不是主动扫描删除。7. 常见问题与故障排查实录在实际操作中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方案SCAN返回的Key数量远小于预期原因ACOUNT参数理解有误。COUNT是每次迭代扫描的“桶”数不是Key数。在Key分布稀疏或模式匹配度低时单次返回的Key可能很少。这是正常的只要最终迭代完能找全就行。原因B迭代过程中数据集发生了较大变化。比如发生了Rehash哈希表扩容。SCAN在Rehash时可能会遗漏或重复部分Key。对于稳定的数据库这种情况较少。对策增加COUNT值并在脚本最后统计扫描到的唯一Key总数用集合存储与一个粗略的估计值如INFO命令中的keyspace信息进行比较。脚本运行中途断开如何续删方案这就是我们之前提到的进度保存。将cursor和deleted_count定期持久化。脚本启动时先检查是否存在进度文件如果存在则读取游标并从那里开始SCAN。注意SCAN的游标是与迭代状态绑定的如果数据库在中断期间发生了结构性变化如大量Key增删导致Rehash从旧游标继续可能不准确但通常问题不大。删除操作导致Redis延迟升高分析DEL命令是O(1)的但释放大量内存时Redis需要整理内存页这个操作特别是如果启用了透明大页THP可能会引起延迟毛刺。对策增加delay参数降低删除频率。分而治之使用max_keys_per_run参数将一次大规模删除拆分成多个小任务在业务低峰期分批执行。监控redis-cli --latency-history观察延迟变化。误删了Key怎么办预防远胜于治疗这就是dry_run和二次确认存在的意义。操作前务必核对。恢复如果没有备份RDB/AOFRedis的数据删除是物理删除无法恢复。如果配置了AOF持久化且没有执行BGREWRITEAOF可以尝试在绝对专业指导下从AOF文件中找回删除命令并手动撤销但这极其复杂且风险高。因此再次强调生产环境操作前先在其他环境测试并务必进行模拟演练和确认。redis-py报ConnectionError或TimeoutError检查网络是否稳定Redis服务器负载是否过高socket_timeout参数是否设置得太短优化增加socket_timeout启用连接池并添加重试机制。可以使用tenacity库为关键操作添加自动重试。from tenacity import retry, stop_after_attempt, wait_fixed retry(stopstop_after_attempt(3), waitwait_fixed(1)) def safe_delete_batch(keys): pipe r.pipeline() for key in keys: pipe.delete(key) return pipe.execute()8. 完整脚本示例与使用指南最后我将给出一个整合了上述所有考量的、相对完整的生产级脚本框架。你可以根据实际情况进行修改和填充。#!/usr/bin/env python3 Redis模糊匹配Key批量删除工具 (生产增强版) 支持模拟扫描、安全删除、进度保存、延迟控制、集群模式。 import os import sys import time import logging import json from typing import Optional, Tuple import redis # 配置区域 REDIS_HOST os.getenv(REDIS_HOST, 127.0.0.1) REDIS_PORT int(os.getenv(REDIS_PORT, 6379)) REDIS_PASSWORD os.getenv(REDIS_PASSWORD, None) REDIS_DB int(os.getenv(REDIS_DB, 0)) IS_CLUSTER os.getenv(REDIS_CLUSTER, false).lower() true # 删除任务参数 PATTERN user:session:* # 要匹配的Key模式 BATCH_SIZE 500 # 每次SCAN的COUNT值 DELAY_SECONDS 0.005 # 每批删除后的延迟秒 MAX_KEYS_PER_RUN 10000 # 单次运行最大删除Key数None表示无限制 PROGRESS_FILE ./redis_deletion_progress.json # 配置结束 def setup_logging(): 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(redis_key_deletion.log), logging.StreamHandler(sys.stdout) ] ) return logging.getLogger(__name__) def get_redis_client(): 获取Redis客户端支持单机和集群模式 if IS_CLUSTER: from redis.cluster import RedisCluster startup_nodes [{host: REDIS_HOST, port: REDIS_PORT}] client RedisCluster( startup_nodesstartup_nodes, passwordREDIS_PASSWORD, decode_responsesTrue, socket_connect_timeout5, socket_timeout10, skip_full_coverage_checkTrue ) else: pool redis.ConnectionPool( hostREDIS_HOST, portREDIS_PORT, passwordREDIS_PASSWORD, dbREDIS_DB, decode_responsesTrue, socket_connect_timeout5, socket_timeout10, max_connections10 ) client redis.Redis(connection_poolpool) try: client.ping() logger.info(Redis连接成功) return client except Exception as e: logger.error(fRedis连接失败: {e}) sys.exit(1) def load_progress(): 从文件加载进度 if os.path.exists(PROGRESS_FILE): try: with open(PROGRESS_FILE, r) as f: data json.load(f) if data.get(pattern) PATTERN: logger.info(f从进度文件恢复: 游标{data.get(cursor)}, 已删除{data.get(deleted, 0)}) return data.get(cursor, 0), data.get(deleted, 0) except Exception as e: logger.warning(f读取进度文件失败: {e}) return 0, 0 def save_progress(cursor, deleted_count): 保存进度到文件 progress { pattern: PATTERN, cursor: cursor, deleted: deleted_count, timestamp: time.time() } try: with open(PROGRESS_FILE, w) as f: json.dump(progress, f, indent2) except Exception as e: logger.error(f保存进度文件失败: {e}) def dry_run(client, pattern, sample_limit50): 模拟执行只扫描不删除 logger.info(f 开始模拟扫描 (模式: {pattern}) ) cursor 0 sampled_keys [] try: while True and len(sampled_keys) sample_limit: cursor, keys client.scan(cursorcursor, matchpattern, count100) for key in keys: if len(sampled_keys) sample_limit: key_type client.type(key) ttl client.ttl(key) sampled_keys.append((key, key_type, ttl)) if cursor 0: break except Exception as e: logger.error(f模拟扫描出错: {e}) return False logger.info(f模拟扫描完成。共发现约 {len(sampled_keys)} 个Key (采样前{sample_limit}个):) for i, (key, key_type, ttl) in enumerate(sampled_keys[:10]): ttl_str f{ttl}s if ttl 0 else 永不过期 if ttl -1 else 已过期 logger.info(f [{i1}] {key} (类型:{key_type}, TTL:{ttl_str})) if len(sampled_keys) 10: logger.info(f ... 以及另外 {len(sampled_keys)-10} 个Key。) if not sampled_keys: logger.warning(未找到任何匹配的Key。请检查模式是否正确。) return False confirm input(f\n!!! 确认要删除所有匹配 {pattern} 的Key吗(输入 DELETE 以继续): ) return confirm.strip().upper() DELETE def safe_delete_keys(client, pattern, start_cursor0, start_deleted0, batch_size100, delay0, max_keysNone): 安全删除Key的核心函数 cursor start_cursor deleted_count start_deleted total_scanned 0 last_save_count deleted_count logger.info(f开始删除任务。起始游标: {cursor}, 起始删除数: {deleted_count}) try: while True: # 1. SCAN 获取一批Key cursor, keys client.scan(cursorcursor, matchpattern, countbatch_size) total_scanned len(keys) if not keys: if cursor 0: logger.info(迭代结束未发现更多匹配的Key。) break # 游标非0但keys为空继续迭代 continue # 2. 检查运行上限 if max_keys and deleted_count len(keys) max_keys: keys_to_delete keys[:max_keys - deleted_count] logger.info(f接近单次上限 {max_keys}本批仅处理 {len(keys_to_delete)} 个。) else: keys_to_delete keys # 3. 执行批量删除 if keys_to_delete: pipe client.pipeline(transactionFalse) for key in keys_to_delete: pipe.delete(key) results pipe.execute() batch_deleted sum(results) deleted_count batch_deleted logger.info(f批次结果: 扫描{len(keys)}个, 处理{len(keys_to_delete)}个, 成功删除{batch_deleted}个。累计删除: {deleted_count}) # 4. 定期保存进度每删除1000个或每批 if deleted_count - last_save_count 1000: save_progress(cursor, deleted_count) last_save_count deleted_count # 5. 检查是否达到上限 if max_keys and deleted_count max_keys: logger.info(f已达到单次运行上限 {max_keys}。) save_progress(cursor, deleted_count) break # 6. 延迟控制 if delay 0: time.sleep(delay) # 7. 检查迭代是否结束 if cursor 0: logger.info(SCAN迭代已全部完成。) break except KeyboardInterrupt: logger.warning(操作被用户中断。) save_progress(cursor, deleted_count) logger.info(f进度已保存。已删除 {deleted_count} 个Key。) except redis.exceptions.RedisError as e: logger.error(fRedis操作错误: {e}) save_progress(cursor, deleted_count) raise e except Exception as e: logger.error(f未知错误: {e}) save_progress(cursor, deleted_count) raise e # 任务完成清理进度文件 if cursor 0: try: os.remove(PROGRESS_FILE) logger.info(任务完成进度文件已清理。) except: pass logger.info(f删除任务结束。总计扫描 {total_scanned} 个Key删除 {deleted_count} 个Key。) return deleted_count def main(): 主函数 global logger logger setup_logging() logger.info(*50) logger.info(f开始Redis Key批量删除任务) logger.info(f模式: {PATTERN}, 批次大小: {BATCH_SIZE}, 延迟: {DELAY_SECONDS}s) logger.info(*50) # 1. 连接Redis client get_redis_client() # 2. 模拟执行与确认 if not dry_run(client, PATTERN): logger.info(用户取消操作或未找到Key。程序退出。) return # 3. 加载进度如有 start_cursor, start_deleted load_progress() if start_cursor ! 0 or start_deleted ! 0: logger.info(f检测到历史进度将从断点继续。) # 4. 执行删除 try: final_deleted safe_delete_keys( clientclient, patternPATTERN, start_cursorstart_cursor, start_deletedstart_deleted, batch_sizeBATCH_SIZE, delayDELAY_SECONDS, max_keysMAX_KEYS_PER_RUN ) logger.info(f任务执行完毕。最终删除Key数量: {final_deleted}) except Exception as e: logger.error(f删除过程发生严重错误: {e}) sys.exit(1) if __name__ __main__: main()使用指南将脚本保存为redis_key_cleaner.py。根据你的环境修改脚本顶部的配置参数尤其是PATTERN、REDIS_HOST、REDIS_PASSWORD等。强烈建议先在测试环境的Redis上运行验证模式匹配是否正确。在生产环境运行前务必使用dry_run功能进行确认。执行脚本python3 redis_key_cleaner.py。脚本支持断点续传。如果中途停止再次运行会自动从上次的进度继续。所有操作日志会同时输出到屏幕和redis_key_deletion.log文件。这个脚本提供了生产环境所需的基本的稳健性、可观测性和可恢复性。你可以在此基础上根据自己业务的特定需求添加更复杂的监控指标上报、报警触发等功能。记住面对生产数据谨慎永远不嫌多。