关于java: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/sripathikr…

网络监控

• 自定义工具，监控进出 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);
 }

如果本文对您有帮忙，欢送 关注 和点赞`，您的反对是我保持创作的能源。

转载请注明出处！