[转帖]Redis 最佳实践（上）

引言

尽管 redis 是一款非常优秀的 NoSQL 数据库，但更重要的是，作为使用者我们应该学会在不同的场景中如何更好的使用它，更大的发挥它的价值。主要可以从这四个方面进行优化：Redis 键值设计、批处理优化、服务端优化、集群配置优化

1. Redis 慢查询日志使用

Redis 提供了慢日志命令的统计功能，它记录了有哪些命令在执行时耗时比较久。

查看 Redis 慢日志之前，你需要设置慢日志的阈值。例如，设置慢日志的阈值为 5 毫秒，并且保留最近 500 条慢日志记录：

# 命令执行耗时超过 5 毫秒，记录慢日志
CONFIG SET slowlog-log-slower-than 5000
# 只保留最近 500 条慢日志
CONFIG SET slowlog-max-len 500

设置完成之后，所有执行的命令如果操作耗时超过了 5 毫秒，都会被 Redis 记录下来。

此时，你可以执行以下命令，就可以查询到最近记录的慢日志：

slowlog len：查询慢查询日志长度
slowlog get [n]：读取 n 条慢查询日志
slowlog reset：清空慢查询列表

127.0.0.1:6379> SLOWLOG get 5
1) 1) (integer) 12691       # 慢日志ID
   2) (integer) 16027264377  # 执行时间戳
   3) (integer) 6989        # 执行耗时(微秒)
   4) 1) "LRANGE"           # 具体执行的命令和参数
      2) "goods_list:100"
      3) "0"
      4) "-1"
2) 1) (integer) 12692
   2) (integer) 16028254247
   3) (integer) 5454
   4) 1) "GET"
      2) "good_info:100"

有可能会导致操作延迟的情况：

经常使用 O (N) 以上复杂度的命令，例如 SORT、SUNION、ZUNIONSTORE 聚合类命令，要花费更多的 CPU 资源
使用 O (N) 复杂度的命令，但 N 的值非常大，Redis 一次需要返回给客户端的数据过多，更多时间花费在数据协议的组装和网络传输过程中。

你可以使用以下方法优化你的业务：

尽量不使用 O (N) 以上复杂度过高的命令，对于数据的聚合操作，放在客户端做
执行 O (N) 命令，保证 N 尽量的小（推荐 N <= 300），每次获取尽量少的数据，让 Redis 可以及时处理返回

2. Redis 键值设计

2.1 优雅的 key 结构

Redis 的 Key 虽然可以自定义，但最好遵循下面的几个最佳实践约定：

遵循基本格式：[业务名称]:[数据名]:[id]
长度不超过 44 字节
不包含特殊字符

例如：我们的登录业务，保存用户信息，其 key 可以设计成如下格式：

这样设计的好处：

可读性强
避免 key 冲突
方便管理
更节省内存： key 是 string 类型，底层编码包含 int、embstr 和 raw 三种。embstr 在小于 44 字节使用，采用连续内存空间，内存占用更小。当字节数大于 44 字节时，会转为 raw 模式存储，在 raw 模式下，内存空间不是连续的，而是采用一个指针指向了另外一段内存空间，在这段空间里存储 SDS 内容，这样空间不连续，访问的时候性能也就会收到影响，还有可能产生内存碎片

2.2 拒绝 BigKey

2.2.1 什么是 BigKey

如果一个 key 写入的 value 非常大，那么 Redis 在分配内存时就会比较耗时。同样的，当删除这个 key 时，释放内存也会比较耗时，这种类型的 key 我们一般称之为 bigkey。

BigKey 通常以 Key 的大小和 Key 中成员的数量来综合判定，例如：

Key 本身的数据量过大：一个 String 类型的 Key ，它的值为 5 MB
Key 中的成员数过多：一个 ZSET 类型的 Key ，它的成员数量为 10,000 个
Key 中成员的数据量过大：一个 Hash 类型的 Key ，它的成员数量虽然只有 1,000 个但这些成员的 Value（值）总大小为 100 MB

那么如何判断元素的大小呢？redis 也给我们提供了命令

MEMORY USAGE KEY

推荐值：

单个 key 的 value 小于 10KB
对于集合类型的 key，建议元素数量小于 1000

2.2.2 BigKey 的危害

网络阻塞

对 BigKey 执行读请求时，少量的 QPS 就可能导致带宽使用率被占满，导致 Redis 实例，乃至所在物理机变慢
数据倾斜

BigKey 所在的 Redis 实例内存使用率远超其他实例，无法使数据分片的内存资源达到均衡
Redis 阻塞

对元素较多的 hash、list、zset 等做运算会耗时较旧，使主线程被阻塞
CPU 压力

对 BigKey 的数据序列化和反序列化会导致 CPU 的使用率飙升，影响 Redis 实例和本机其它应用

2.2.3 如何发现 BigKey

redis-cli --bigkeys  -a `密码`

利用 redis-cli 提供的–bigkeys 参数，可以遍历分析所有 key，并返回 Key 的整体统计信息与每个数据类型的 Top1 的 big key

这个命令的原理，就是 Redis 在内部执行了 SCAN 命令，遍历整个实例中所有的 key，然后针对 key 的类型，分别执行 STRLEN、LLEN、HLEN、SCARD、ZCARD 命令，来获取 String 类型的长度、容器类型（List、Hash、Set、ZSet）的元素个数。

这里需要提醒你的是，当执行这个命令时，要注意 2 个问题：

对线上实例进行 bigkey 扫描时，Redis 的 OPS 会突增，为了降低扫描过程中对 Redis 的影响，最好控制一下扫描的频率，指定 -i 参数即可，它表示扫描过程中每次扫描后休息的时间间隔，单位是秒
扫描结果中，对于容器类型（List、Hash、Set、ZSet）的 key，只能扫描出元素最多的 key。但一个 key 的元素多，不一定表示占用内存也多，你还需要根据业务情况，进一步评估内存占用情况

scan cursor count n

自己编程，利用 scan 扫描 Redis 中的所有 key，利用 strlen、hlen 等命令判断 key 的长度（此处不建议使用 MEMORY USAGE）

scan 命令调用完后每次会返回 2 个元素，第一个是下一次迭代的光标，第一次光标会设置为 0，当最后一次 scan 返回的光标等于 0 时，表示整个 scan 遍历结束了，第二个返回的是 List，一个匹配的 key 的数组

public class JedisTest {
    private Jedis jedis;
    @BeforeEach
    void setUp() {
        // 1.建立连接
        // jedis = new Jedis("192.168.150.101", 6379);
        jedis = JedisConnectionFactory.getJedis();
        // 2.设置密码
        jedis.auth("123321");
        // 3.选择库
        jedis.select(0);
    }
    final static int STR_MAX_LEN = 10 * 1024;
    final static int HASH_MAX_LEN = 500;
    @Test
    void testScan() {
        int maxLen = 0;
        long len = 0;
        String cursor = "0";
        do {
            // 扫描并获取一部分key
            ScanResult<String> result = jedis.scan(cursor);
            // 记录cursor
            cursor = result.getCursor();
            List<String> list = result.getResult();
            if (list == null || list.isEmpty()) {
                break;
            }
            // 遍历
            for (String key : list) {
                // 判断key的类型
                String type = jedis.type(key);
                switch (type) {
                    case "string":
                        len = jedis.strlen(key);
                        maxLen = STR_MAX_LEN;
                        break;
                    case "hash":
                        len = jedis.hlen(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "list":
                        len = jedis.llen(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "set":
                        len = jedis.scard(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "zset":
                        len = jedis.zcard(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    default:
                        break;
                }
                if (len >= maxLen) {
                    System.out.printf("Found big key : %s, type: %s, length or size: %d %n", key, type, len);
                }
            }
        } while (!cursor.equals("0"));
    }
    @AfterEach
    void tearDown() {
        if (jedis != null) {
            jedis.close();
        }
    }
}

第三方工具

利用第三方工具，如 Redis-Rdb-Tools 分析 RDB 快照文件，全面分析内存使用情况
https://github.com/sripathikrishnan/redis-rdb-tools

网络监控

自定义工具，监控进出 Redis 的网络数据，超出预警值时主动告警
一般阿里云搭建的云服务器就有相关监控页面

2.2.4 BigKey 解决方案

这里有两点可以优化：

业务应用尽量避免写入 bigkey
如果你使用的 Redis 是 4.0 以上版本，用 UNLINK 命令替代 DEL，此命令可以把释放 key 内存的操作，放到后台线程中去执行，从而降低对 Redis 的影响
如果你使用的 Redis 是 6.0 以上版本，可以开启 lazy-free 机制（lazyfree-lazy-user-del = yes），在执行 DEL 命令时，释放内存也会放到后台线程中执行

bigkey 在很多场景下，都会产生性能问题。例如，bigkey 在分片集群模式下，对于数据的迁移也会有性能影响，以及我后面即将讲到的数据过期、数据淘汰、透明大页，都会受到 bigkey 的影响。因此，即使 reids6.0 以后，仍然不建议使用 BigKey

2.3 总结

Key 的最佳实践
- 固定格式：[业务名]:[数据名]:[id]
- 足够简短：不超过 44 字节
- 不包含特殊字符
Value 的最佳实践：
- 合理的拆分数据，拒绝 BigKey
- 选择合适数据结构
- Hash 结构的 entry 数量不要超过 1000
- 设置合理的超时时间

3. 批处理优化

3.1 Pipeline

3.1.1 客户端与服务端交互

单个命令的执行流程

N 条命令的执行流程

redis 处理指令是很快的，主要花费的时候在于网络传输。于是乎很容易想到将多条指令批量的传输给 redis

3.1.2 MSet

Redis 提供了很多 Mxxx 这样的命令，可以实现批量插入数据，例如：

mset
hmset

利用 mset 批量插入 10 万条数据

@Test
void testMxx() {
    String[] arr = new String[2000];
    int j;
    long b = System.currentTimeMillis();
    for (int i = 1; i <= 100000; i++) {
        j = (i % 1000) << 1;
        arr[j] = "test:key_" + i;
        arr[j + 1] = "value_" + i;
        if (j == 0) {
            jedis.mset(arr);
        }
    }
    long e = System.currentTimeMillis();
    System.out.println("time: " + (e - b));
}

3.1.3 Pipeline

MSET 虽然可以批处理，但是却只能操作部分数据类型，因此如果有对复杂数据类型的批处理需要，建议使用 Pipeline

@Test
void testPipeline() {
    // 创建管道
    Pipeline pipeline = jedis.pipelined();
    long b = System.currentTimeMillis();
    for (int i = 1; i <= 100000; i++) {
        // 放入命令到管道
        pipeline.set("test:key_" + i, "value_" + i);
        if (i % 1000 == 0) {
            // 每放入1000条命令，批量执行
            pipeline.sync();
        }
    }
    long e = System.currentTimeMillis();
    System.out.println("time: " + (e - b));
}

3.2 集群下的批处理

如 MSET 或 Pipeline 这样的批处理需要在一次请求中携带多条命令，而此时如果 Redis 是一个集群，那批处理命令的多个 key 必须落在一个插槽中，否则就会导致执行失败。大家可以想一想这样的要求其实很难实现，因为我们在批处理时，可能一次要插入很多条数据，这些数据很有可能不会都落在相同的节点上，这就会导致报错了

这个时候，我们可以找到 4 种解决方案

第一种方案：串行执行，所以这种方式没有什么意义，当然，执行起来就很简单了，缺点就是耗时过久。
第二种方案：串行 slot，简单来说，就是执行前，客户端先计算一下对应的 key 的 slot ，一样 slot 的 key 就放到一个组里边，不同的，就放到不同的组里边，然后对每个组执行 pipeline 的批处理，他就能串行执行各个组的命令，这种做法比第一种方法耗时要少，但是缺点呢，相对来说复杂一点，所以这种方案还需要优化一下
第三种方案：并行 slot，相较于第二种方案，在分组完成后串行执行，第三种方案，就变成了并行执行各个命令，所以他的耗时就非常短，但是实现呢，也更加复杂。
第四种：hash_tag，redis 计算 key 的 slot 的时候，其实是根据 key 的有效部分来计算的，通过这种方式就能一次处理所有的 key，这种方式耗时最短，实现也简单，但是如果通过操作 key 的有效部分，那么就会导致所有的 key 都落在一个节点上，产生数据倾斜的问题，所以我们推荐使用第三种方式。

3.2.1 串行化执行代码实践

public class JedisClusterTest {

    private JedisCluster jedisCluster;

    @BeforeEach
    void setUp() {
        // 配置连接池
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(8);
        poolConfig.setMaxIdle(8);
        poolConfig.setMinIdle(0);
        poolConfig.setMaxWaitMillis(1000);
        HashSet<HostAndPort> nodes = new HashSet<>();
        nodes.add(new HostAndPort("192.168.150.101", 7001));
        nodes.add(new HostAndPort("192.168.150.101", 7002));
        nodes.add(new HostAndPort("192.168.150.101", 7003));
        nodes.add(new HostAndPort("192.168.150.101", 8001));
        nodes.add(new HostAndPort("192.168.150.101", 8002));
        nodes.add(new HostAndPort("192.168.150.101", 8003));
        jedisCluster = new JedisCluster(nodes, poolConfig);
    }

    @Test
    void testMSet() {
        jedisCluster.mset("name", "Jack", "age", "21", "sex", "male");

    }

    @Test
    void testMSet2() {
        Map<String, String> map = new HashMap<>(3);
        map.put("name", "Jack");
        map.put("age", "21");
        map.put("sex", "Male");
        //对Map数据进行分组。根据相同的slot放在一个分组
        //key就是slot，value就是一个组
        Map<Integer, List<Map.Entry<String, String>>> result = map.entrySet()
                .stream()
                .collect(Collectors.groupingBy(
                        entry -> ClusterSlotHashUtil.calculateSlot(entry.getKey()))
                );
        //串行的去执行mset的逻辑
        for (List<Map.Entry<String, String>> list : result.values()) {
            String[] arr = new String[list.size() * 2];
            int j = 0;
            for (int i = 0; i < list.size(); i++) {
                j = i<<2;
                Map.Entry<String, String> e = list.get(0);
                arr[j] = e.getKey();
                arr[j + 1] = e.getValue();
            }
            jedisCluster.mset(arr);
        }
    }

    @AfterEach
    void tearDown() {
        if (jedisCluster != null) {
            jedisCluster.close();
        }
    }
}

3.2.2 Spring 集群环境下批处理代码

@Test
 void testMSetInCluster() {
     Map<String, String> map = new HashMap<>(3);
     map.put("name", "Rose");
     map.put("age", "21");
     map.put("sex", "Female");
     stringRedisTemplate.opsForValue().multiSet(map);
     List<String> strings = stringRedisTemplate.opsForValue().multiGet(Arrays.asList("name", "age", "sex"));
     strings.forEach(System.out::println);
 }

本文由传智教育博学谷教研团队发布。

如果本文对您有帮助，欢迎关注和点赞；如果您有任何建议也可留言评论或私信，您的支持是我坚持创作的动力。

转载请注明出处！