需求

需要将一个redis实例中(或是具体到某一个db)的部分keys，转移到另一个redis实例(或是具体到某一个db)

使用Redis自身支持的指令

源实例与目标实例版本相同

使用dump命令

#!/bin/bash
#redis 源ip
src_ip=127.0.0.1
#redis 源port
src_port=6392
#redis 源密码
src_auth=
#redis 源库
src_db=

#redis 目的ip
dest_ip=127.0.0.1
#redis 目的port
dest_port=6393
#redis 目的密码
dest_auth=
#redis 目的库
dest_db=

#要迁移的key前缀
key_prefix=test

i=1

redis-cli -h $src_ip -p $src_port -a $src_auth -n $src_db keys "${key_prefix}*" | while read key
do
    redis-cli -h $dest_ip -p $dest_port -a $dest_auth -n $dest_db del $key
    redis-cli -h $src_ip -p $src_port -a $src_auth -n $src_db --raw dump $key | perl -pe 'chomp if eof' | redis-cli -h $dest_ip -p $dest_port -a $dest_auth -n $dest_db -x restore $key 0
    echo "$i migrate key $key"
    i=$(( $i + 1 ))
done

使用migrate命令

1	migrate host port key\|"" auth {password} destination-db timeout [copy] [replace] [keys...]

migrate命令也是用于在Redis实例间进行数据迁移的，实际上migrate命令就是将dump、 restore、 del三个命令进行组合，从而简化了操作流程。migrate命令具有原子性，而且从Redis3.0.6版本以后已经支持迁移多个键的功能，有效地提高了迁移效率， migrate在10.4节水平扩容中起到重要作用。
3.0之后提供的auth参数

实现过程和dump+restore基本类似，但是有3点不太相同：

第一，整个过程是原子执行的，不需要在多个Redis实例上开启客户端的，只需要在源Redis上执行migrate命令即可。

第二， migrate命令的数据传输直接在源Redis和目标Redis上完成的。

第三，目标Redis完成restore后会发送OK给源Redis，源Redis接收后会根据migrate对应的选项来决定是否在源Redis上删除对应的键。

下面对migrate的参数进行逐个说明：

host：目标Redis的IP地址。
port：目标Redis的端口。
key|""：在Redis3.0.6版本之前， migrate只支持迁移一个键，所以此处是要迁移的键，但Redis3.0.6版本之后支持迁移多个键，如果当前需要迁移多个键，此处为空字符串""。
destination-db：目标Redis的数据库索引，例如要迁移到0号数据库，这里就写0。
timeout：迁移的超时时间（单位为毫秒）。
[copy]：如果添加此选项，迁移后并不删除源键。
[replace]：如果添加此选项， migrate不管目标Redis是否存在该键都会正常迁移进行数据覆盖。
[keys key[key...]]：迁移多个键，例如要迁移key1、 key2、 key3，此处填写“keys key1 key2 key3”。

1
2
3

127.0.0.1:6379> migrate 127.0.0.1 6380 hello 0 1000
127.0.0.1:6379> migrate 127.0.0.1 6379 hello 0 1000 replace
127.0.0.1:6379> migrate 127.0.0.1 6380 "" 0 5000 keys key1 key2 key3

迁移脚本

#!/bin/bash

#redis 源ip
src_ip=127.0.0.1
#redis 源port
src_port=6392
#redis 源密码
src_auth=
#redis 源库
src_db=

#redis 目的ip
dest_ip=127.0.0.1
#redis 目的port
dest_port=6393
#redis 目的密码
dest_auth=
#redis 目的库
dest_db=


#要迁移的key前缀
key_prefix=test

i=1

redis-cli -h $src_ip -p $src_port -a $src_auth -n $src_db keys "${key_prefix}*" | while read key
do
    redis-cli -h $src_ip -p $src_port -a $src_auth -n $src_db migrate $dest_ip $dest_port $key $dest_db 1000 auth $dest_auth replace
    echo "$i migrate key $key"
    i=$(( $i + 1 ))
done

源实例与目标实例版本不相同

不可行方案

如果源实例与目标实例版本不相同，使用migrate进行迁移的时候会有如下错误：

1 2	1935 migrate key esf_common_auth_code_18587656289 (error) ERR Target instance replied with error: ERR DUMP payload version or checksum are wrong

如果源实例与目标实例版本不相同，使用dump进行迁移的时候会有如下错误

1	(error) ERR DUMP payload version or checksum are wrong

如果源实例与目标实例版本不相同,直接复制rdbw文件搭建从库为有如下报错

5453:S 23 Nov 18:13:14.153 * MASTER <-> SLAVE sync: Flushing old data
5453:S 23 Nov 18:13:14.153 * MASTER <-> SLAVE sync: Loading DB in memory
5453:S 23 Nov 18:13:14.153 # Can't handle RDB format version 8
5453:S 23 Nov 18:13:14.153 # Failed trying to load the MASTER synchronization DB from disk

可行方案

开启源实例aof持久化功能

1	config set appendonly yes

手动进行aof持久化

1	bgrewriteaof

新建一个redis实例，与目标实例版本相同并启动

将源实例的redis的aof文件导入新建实例

1	redis-cli -h 127.0.0.1 -p 6395 -a password --pipe < appendonly.aof

通过dump或者migrate的方式将新实例的key迁移到目标实例

RedisShake(第三方工具 - 无需考虑版本且提供了增量同步方案)

注：
当源实例为腾讯云Redis时，可能遇到ERR ERR psync is disabled command错误，是因为腾讯云本身不支持psync 指令
增量同步需要保证源实例开启了aof

Python脚本迁移(无需考虑是否支持psync指令以及版本)

安装redis：pip3 install redis -i https://pypi.tuna.tsinghua.edu.cn/simple/
执行脚本：python3 redis-migrate.py

import redis

source_db = 9
target_db = 13

# 从source_redis迁移数据到target_redis
source_redis = redis.Redis(host='xxx', port=6379, db=source_db, password="xxx")
target_redis = redis.Redis(host='xxx', port=6379, db=target_db, password="xxx")

# 通配扫描
pattern = '*'

cursor = '0'
# 每次扫描多少
count = 100
while cursor != 0:
    cursor, keys = source_redis.scan(cursor=cursor, match=pattern, count=count)
    pipe = target_redis.pipeline(transaction=False)
    for key in keys:
        print(f'开始执行-》{source_db}:{str(key, encoding="utf-8")}')
        t = source_redis.type(key)
        ttl = source_redis.ttl(key)
        if t == b'string':
            pipe.set(key, source_redis.get(key))
        elif t == b'hash':
            hash_data = source_redis.hgetall(key)
            pipe.hset(key, mapping=hash_data)
        elif t == b'list':
            pipe.rpush(key, *source_redis.lrange(key, 0, -1))
        elif t == b'set':
            pipe.sadd(key, *source_redis.smembers(key))
        elif t == b'zset':
            # 使用 withscores=True 从源 Redis 获取成员和分数,构造成字典映射,然后传递给 zadd 的 mapping 参数。可以避免参数数量过多的问题（TypeError: SortedSetCommands.zadd() takes from 3 to 9 positional arguments but xxx were give）
            score_member_mapping = {
                score: member
                for score, member in source_redis.zrange(key, 0, -1, withscores=True)
            }
            pipe.zadd(key, mapping=score_member_mapping)
        else:
            # 其他类型的数据处理
            pass
        if ttl > 0:
            # 将源key的TTL也一并迁移过来
            pipe.expire(key, ttl)
    pipe.execute()

删除(按指定DB和key-prefix删除)

场景一：删除所有的 key

如果需要执行初始化的操作，清理掉数据库所有的键，可以使用 FLUSHDB 或者 FLUSHALL 命令操作。

1 2	FLUSHDB：删除当前数据库中的所有 key 。 FLUSHALL：删除当前连接所有数据库的所有 key 。

场景二：删除所有满足匹配条件的 key（ key 数量较少或者测试环境）

可以在命令行环境下使用 redis-cli 命令在外部执行 KEYS {pattern} 命令，拿到结果以后通过 xargs 命令传递给 DEL 作为输入参数，进而删除匹配的 key 。具体命令如下：

1	redis-cli -h {hostname} -p {port} -a {password} -n {database} --raw keys "{pattern}" \| xargs -I {} redis-cli -h {hostname} -p {port} -a {password} -n {database} DEL "{}"

说明：
redis-cli 是访问 Redis 的客户端命令，用法是：redis-cli [OPTIONS] [cmd [arg [arg ...]]]。
hostname：服务器主机，port：服务器端口，a：密码（无密码可缺省），n：数据库编号。
低版本的 Redis 会在返回结果中加上数字编号，使用 --raw 参数可以去掉结果编号。
xargs -I {} 参数可以避免 key 中存在空格导致的参数拆分异常问题

但是这种操作是有限制的，主要受限于 KEYS 命令。因为 Redis 6 版本以下都是采用单线程处理请求，如果在 key 数量较大的情况下使用 KEYS 命令，会阻塞线程，导致其他客户端无法正常访问，这在生产环境是不可接受的（基于此原因，很多公司生产环境 KEYS 都是禁用的）。这就是接下来要说的第三种场景。

场景三：删除所有满足匹配条件的 key（ key 数量较多或者生产环境 - Scan）

为了解决场景二中的 KEYS 命令造成的线程阻塞问题，我们可以使用 SCAN 命令来解决。

让我们先来了解一下 SCAN 命令的使用。

SCAN 用于迭代当前数据库中的数据库键，用法如下：

1	SCAN cursor [MATCH pattern] [COUNT count]

简单概括一下： SCAN 命令就是通过游标的方式分步从数据库获取数据，每次以游标方式进行遍历（游标从上一次遍历结果中返回，初始游标为 0 ），结果会返回一个新游标和匹配的键集合（返回键的数量不不确定，小于等于 COUNT ），如果返回游标为 0 则视为遍历结束（不以遍历结果为空作为结束标识）。可以使用 MATCH 参数匹配模式，COUNT 参数限制返回的键的个数。

与 KEYS 命令相比，SCAN 命令虽然复杂度也是 O(n)，但是它是通过游标分步进行的，不会阻塞线程。同时 redis-cli 命令本身支持 --scan 和 --pattern 的参数，可以直接在命令行获取到匹配的结果。

修改后的命令如下：

1	redis-cli -h {hostname} -p {port} -a {password} -n {database} --raw --scan --pattern "{pattern}" \| xargs -L 100 -I {} redis-cli -h {hostname} -p {port} -a {password} -n {database} DEL "{}"

注意这里并没有直接在 redis-cli 中使用 SCAN 命令，而是用 --scan 参数的方式调用，这是因为 SCAN 命令会返回两个参数（游标和结果），这不利于作为 xargs 的输入参数，而 --scan 参数只返回匹配的键，可以和 xargs 命令完美结合。而且可以给 xargs 指定输入参数的条数（-L），进一步限制每一次删除的键的个数（在没有指定 COUNT 参数情况下，默认值是 10 ，SCAN 每次会返回最多 10 个左右的数据，并非严格相等）。

到这里看似问题已经得到了解决，但是在实际场景中这种处理方式还是存在一些问题。

SCAN 命令虽然解决了线程阻塞的问题，但是也带来了效率的问题。假设数据库 key 的数量级在 10w+ 左右，需要删除的 key 数量级在 100+ 左右，这时候如果使用上述命令手动操作的话无疑是十分痛苦的，需要不断地重复执行（ SCAN 命令并不是每次都可以返回匹配到的结果集，只要没有返回 0 游标，就需要继续遍历）。另外一次给 DEL 传递过多的参数也不是一种很好的选择，因为如果 key 比较大时，使用 DEL 删除本身也会造成线程阻塞，这样整个命令的阻塞时间就取决于 key 的数量和大小。

综上所述，主要有两个影响因素需要考虑：一个是如何在不阻塞线程的情况下，高效查询匹配的 key ；另一个是如何避免在执行删除操作的时候造成线程阻塞。

针对第一种情况，可以考虑通过脚本程序执行 SCAN 命令，这样就不必担心重复执行的效率问题了。针对 DEL 命令可能造成的阻塞问题，可以使用 EXPIRE 命令替换。以 PHP 语言为例，可以使用以下脚本进行处理：

use Predis\Client;

$client = new Client();
while (1) {
    list($iterator, $result) = $client->scan(0, ['MATCH' => 'PHP*', 'COUNT' => 50]);
    foreach ($result as $key) {
        $client->expire($key, mt_rand(0, 600));
    }
    if ($iterator == "0") {
        break;
    }
}

说明：
这里使用的是 predis/predis composer 包，安装方式：composer require predis/predis。
不使用遍历返回的结果集作为 while 的判断条件是因为 SCAN 命令结束的标志是返回值为 0 的游标。
使用 EXPIRE 设置过期的时候，过期时间采用了随机数的方式，是为了防止在删除 key 的数量过多时，同一时间集中过期引起雪崩现象。

当然，如果需要删除的 key 和 key 的总数数量级相差太大的话，使用 SCAN 命令遍历的效率还是差了些。这时可以借助 BGSAVE 生成 rdb 文件，然后再通过 rdb分析工具（rdbtools） 获取需要操作的 key ，借助程序进行过期处理。这个小技巧我们会在介绍「rdb 文件」的应用场景的时候单独介绍。