生产环境中要禁止使用keys *查询key, 因为redis低版本是单线程，如果key非常多的话，直接使用keys *会导致阻塞，所以应当使用scan命令，scan命令介绍请参考其他文档。
python"># -*- coding: utf-8 -*-
# @Time : 2025/01/09
# @Author : 养乐多
# @Mail : oct_hao@163.com
# @File : redis_del_keys.py
# @Use : 利用scan命令批量删除redis keyimport redisclass Redis:def __init__(self, host, port, db=0):'''初始化redis连接:param host: redis host:param port: redis port:param db: redis db'''self.redis_host = hostself.redis_port = portself.redis_db = dbtry:# 连接redisself.redis_client = self._connect()except Exception as e:print('redis连接失败', e)def _connect(self):'''连接redis'''return redis.StrictRedis(host=self.redis_host, port=self.redis_port, db=self.redis_db)def _close(self):'''关闭redis连接'''self.redis_client.close()def batch_delete_keys(self, pattern, batch_size=1000):"""使用 SCAN 命令批量删除匹配给定模式的 Redis 键:param pattern: 要删除的键的模式:param batch_size: 每次扫描时的键的批量大小"""# 初始化游标cursor = 0 while True:# 使用 SCAN 命令遍历键cursor, keys = self.redis_client.scan(cursor=cursor, match=pattern, count=batch_size)# 删除匹配的键if keys:self.redis_client.delete(*keys)print('删除了%s个key' % len(keys))else:breakif cursor == 0:breakself._close()def get_all_key_count(self, pattern, batch_size=1000):'''使用 SCAN 命令批量删除匹配给定模式的 Redis 键:param pattern: 要删除的键的模式:param batch_size: 每次扫描时的键的批量大小'''# 初始化游标cursor = 0# 初始化计数器count = 0 while True:# 使用 SCAN 命令遍历键cursor, keys = self.redis_client.scan(cursor=cursor, match=pattern, count=batch_size)# 统计匹配的键的数量if keys:count += len(keys)else:breakif cursor == 0:breakself._close()return countif __name__ == "__main__":redis_host = '172.30.11.132'redis_port = 6379redis_db = 0pattern = '*123'batch_size = 1000# 连接到 Redisr = Redis(redis_host, redis_port, redis_db)# 查询key总数print(r.get_all_key_count(pattern=pattern, batch_size=batch_size))# 删除key# r.batch_delete_keys(pattern=pattern, batch_size=batch_size)
