【最佳实践】- Redis

概述

本文介绍如何玩转 Redis, 可以说是 Redis 开发规范, 也可以理解为 Redis 最佳实战.

一、键值设计

1. key名 设计

(1). 【强制】: 可读性和可管理性

以业务名(或数据库名)为前缀(防止key冲突), 用冒号(句号)分隔, 比如: 业务名:表名:id

• csdn:user:1

(2). 【建议】: 简洁性

保证语义的前提下, 控制 key 的长度, 当 key 较多时, 内存占用也不容忽视, 例如:

• user:{uid}:friends:messages:{mid} 简化为 u:{uid}🇫🇷m:{mid}

(3). 【强制】: 不要包含特殊字符

反例: 包含 空格 、换行 、单双引号 以及 其他转义字符

2. value 设计

(1). 【强制】: 拒绝 bigkey(防止网卡流量、慢查询)

说明: 非字符串的bigkey，不要使用del删除，使用hscan、sscan、zscan方式渐进式删除，同时要注意防止bigkey过期时间自动删除问题(例如一个200万的zset设置1小时过期，会触发del操作，造成阻塞，而且该操作不会不出现在慢查询中(latency可查))

正例: string 类型控制在 10KB 以内，hash、list、set、zset元素个数不要超过5000
反例: 一个包含 200万 个元素的 list

(2). 【推荐】: 选择适合的数据类型

说明: 实体类型(要合理控制和使用数据结构 内存编码 优化配置, 例如 ziplist，但也要注意节省内存和性能之间的平衡)

正例:

• hmset user:1 name tom age 19 favor football

反例:

• set user:1:name tom
• set user:1:age 19
• set user:1:favor football

(3). 【推荐】: 控制key的生命周期，redis不是垃圾桶

说明: 建议使用 expire 设置过期时间(条件允许可以打散过期时间，防止集中过期)，不过期的数据重点关注 idletime

二、命令使用

(1). 【推荐】: O(N)命令关注N的数量

说明: 例如 hgetall、lrange、smembers、zrange、sinter 等并非不能使用，但是需要 明确N 的值。有 遍历 的需求可以使用 hscan、sscan、zscan 代替

(2). 【推荐】: 禁用命令

说明: 禁止线上使用 keys、flushall、flushdb等，通过 redis 的 rename 机制禁掉命令，或者使用 scan 的方式渐进式处理

(3). 【推荐】: 合理使用select

说明: redis 的多数据库较弱，使用数字进行区分，很多客户端支持较差，同时多业务用多数据库实际还是单线程处理，会有干扰

(4). 【推荐】: 使用批量操作提高效率

• 原生命令：例如 mget、mset
• 非原生命令：可以使用 pipeline 提高效率

说明: 但要注意控制一次批量操作的元素个数(例如500以内，实际也和元素字节数有关)

注意两者不同：

1. 原生是原子操作，pipeline 是非原子操作。
2. pipeline 可以打包不同的命令，原生做不到
3. pipeline 需要客户端和服务端同时支持。

(5). 【建议】: Redis事务功能较弱，不建议过多使用

说明: Redis 的事务功能较弱(不支持回滚)，而且集群版本(自研和官方)要求一次事务操作的 key 必须在一个 slot 上(可以使用hashtag功能解决)

(6). 【建议】: Redis集群版本在使用Lua上有特殊要求

1. 所有 key 都应该由 KEYS 数组来传递，redis.call/pcall 里面调用的 redis 命令，key 的位置，必须是KEYS array, 否则直接返回 error，"-ERR bad lua script for redis cluster, all the keys that the script uses should be passed using the KEYS array"
2. 所有 key，必须在1个 slot 上，否则直接返回error, “-ERR eval/evalsha command keys must in same slot”

(7).【建议】必要情况下使用monitor命令时，要注意不要长时间使用

三、客户端使用

(1). 【推荐】: 避免多个应用使用一个Redis实例

正例: 不相干 的 业务拆分，公共数据 做 服务化

(2). 【推荐】: 使用带有连接池的数据库，可以有效控制连接，同时提高效率，标准使用方式

代码如下:

Jedis jedis = null;
	try {
	    jedis = jedisPool.getResource();
	    //具体的命令
	    jedis.executeCommand()
	} catch (Exception e) {
	    logger.error("op key {} error: " + e.getMessage(), key, e);
	} finally {
	    //注意这里不是关闭连接，在JedisPool模式下，Jedis会被归还给资源池。
	    if (jedis != null) 
	        jedis.close();
	}

(3). 【建议】: 高并发下建议客户端添加熔断功能(例如netflix hystrix)

(4). 【推荐】: 设置合理的密码，如有必要可以使用SSL加密访问

(5). 【推荐】: 根据自身业务类型，选好maxmemory-policy(最大内存淘汰策略)，设置好过期时间

默认策略是 volatile-lru，即超过最大内存后，在过期键中使用 lru 算法进行 key 的剔除，保证不过期数据不被删除，但是可能会出现 OOM 问题。

其他策略如下：

• allkeys-lru：根据 LRU 算法删除键，不管数据有没有设置超时属性，直到腾出足够空间为止。
• allkeys-random：随机删除所有键，直到腾出足够空间为止。
• volatile-random:随机删除过期键，直到腾出足够空间为止。
• volatile-ttl：根据键值对象的 ttl 属性，删除最近将要过期数据。如果没有，回退到 noeviction 策略。
• noeviction：不会剔除任何数据，拒绝所有写入操作并返回客户端错误信息"(error) OOM command not allowed - when used memory"，此时 Redis 只响应读操作。

四、相关工具

(1). 【推荐】: 数据同步

说明: redis 间数据同步可以使用：redis-port

(2). 【推荐】: big key搜索

redis大key搜索工具

(3). 【推荐】: 热点key寻找(内部实现使用monitor，所以建议短时间使用)

facebook的redis-faina

五 附录：删除 bigkey

1. 下面操作可以使用 pipeline 加速
2. redis 4.0已经支持 key 的 异步删除，欢迎使用

(1). Hash删除: hscan + hdel

public void delBigHash(String host, int port, String password, String bigHashKey) {
    Jedis jedis = new Jedis(host, port);
    if (password != null && !"".equals(password)) {
        jedis.auth(password);
    }
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<Entry<String, String>> scanResult = jedis.hscan(bigHashKey, cursor, scanParams);
        List<Entry<String, String>> entryList = scanResult.getResult();
        if (entryList != null && !entryList.isEmpty()) {
            for (Entry<String, String> entry : entryList) {
                jedis.hdel(bigHashKey, entry.getKey());
            }
        }
        cursor = scanResult.getStringCursor();
    } while (!"0".equals(cursor));
    
    //删除bigkey
    jedis.del(bigHashKey);
}

(2). List删除: ltrim

public void delBigList(String host, int port, String password, String bigListKey) {
    Jedis jedis = new Jedis(host, port);
    if (password != null && !"".equals(password)) {
        jedis.auth(password);
    }
    long llen = jedis.llen(bigListKey);
    int counter = 0;
    int left = 100;
    while (counter < llen) {
        //每次从左侧截掉100个
        jedis.ltrim(bigListKey, left, llen);
        counter += left;
    }
    //最终删除key
    jedis.del(bigListKey);
}

(3). Set删除: sscan + srem

public void delBigSet(String host, int port, String password, String bigSetKey) {
    Jedis jedis = new Jedis(host, port);
    if (password != null && !"".equals(password)) {
        jedis.auth(password);
    }
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<String> scanResult = jedis.sscan(bigSetKey, cursor, scanParams);
        List<String> memberList = scanResult.getResult();
        if (memberList != null && !memberList.isEmpty()) {
            for (String member : memberList) {
                jedis.srem(bigSetKey, member);
            }
        }
        cursor = scanResult.getStringCursor();
    } while (!"0".equals(cursor));
    
    //删除bigkey
    jedis.del(bigSetKey);
}

(4). SortedSet删除: zscan + zrem

public void delBigZset(String host, int port, String password, String bigZsetKey) {
    Jedis jedis = new Jedis(host, port);
    if (password != null && !"".equals(password)) {
        jedis.auth(password);
    }
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<Tuple> scanResult = jedis.zscan(bigZsetKey, cursor, scanParams);
        List<Tuple> tupleList = scanResult.getResult();
        if (tupleList != null && !tupleList.isEmpty()) {
            for (Tuple tuple : tupleList) {
                jedis.zrem(bigZsetKey, tuple.getElement());
            }
        }
        cursor = scanResult.getStringCursor();
    } while (!"0".equals(cursor));
    
    //删除bigkey
    jedis.del(bigZsetKey);
}

最佳实践

热点key问题(四种解决方案)

1. 使用互斥锁

如果是单机，可以用 synchronized 或者 lock 来处理，如果是 分布式 环境可以用 分布式锁 就可以了（分布式锁，可以用 memcache 的add, redis 的setnx, zookeeper 的添加节点 操作）。

memcache 的伪代码实现

if (memcache.get(key) == null) {
    // 3 min timeout to avoid mutex holder crash
    if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {
        value = db.get(key);
        memcache.set(key, value);
        memcache.delete(key_mutex);
    } else {
        sleep(50);
        retry();
    }
}

redis 的伪代码实现

String get(String key) {
   String value = redis.get(key);
   if (value  == null) {
    if (redis.setnx(key_mutex, "1")) {
        // 3 min timeout to avoid mutex holder crash
        redis.expire(key_mutex, 3 * 60)
        value = db.get(key);
        redis.set(key, value);
        redis.delete(key_mutex);
    } else {
        //其他线程休息50毫秒后重试
        Thread.sleep(50);
        get(key);
    }
  }

2. "提前"使用互斥锁

在 value 内部设置1个超时值(timeout1), timeout1 比实际的 memcache timeout(timeout2) 小。当从 cache 读取到 timeout1 发现它已经过期时候，马上 延长timeout1 并重新设置到 cache。然后再从数据库加载数据并设置到 cache 中。

伪代码如下：

v = memcache.get(key);
if (v == null) {
    if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {
        value = db.get(key);
        memcache.set(key, value);
        memcache.delete(key_mutex);
    } else {
        sleep(50);
        retry();
    }
} else {
    if (v.timeout <= now()) {
        if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {
            // extend the timeout for other threads
            v.timeout += 3 * 60 * 1000;
            memcache.set(key, v, KEY_TIMEOUT * 2);

            // load the latest value from db
            v = db.get(key);
            v.timeout = KEY_TIMEOUT;
            memcache.set(key, value, KEY_TIMEOUT * 2);
            memcache.delete(key_mutex);
        } else {
            sleep(50);
            retry();
        }
    }
}

3. 永远不过期

这里的“永远不过期”包含两层意思：

1. 从redis上看，确实没有设置过期时间，这就保证了，不会出现热点key过期问题，也就是“物理”不过期。

2. 从功能上看，如果不过期，那不就成静态的了吗？所以我们把过期时间存在key对应的value里，如果发现要过期了，通过一个后台的异步线程进行缓存的构建，也就是“逻辑”过期

从实战看，这种方法对于性能非常友好，唯一不足的就是构建缓存时候，其余线程(非构建缓存的线程)可能访问的是老数据，但是对于一般的互联网功能来说这个还是可以忍受。

String get(final String key) {
    V v = redis.get(key);
    String value = v.getValue();
    long timeout = v.getTimeout();
    if (v.timeout <= System.currentTimeMillis()) {
        // 异步更新后台异常执行
        threadPool.execute(new Runnable() {
            public void run() {
                String keyMutex = "mutex:" + key;
                if (redis.setnx(keyMutex, "1")) {
                    // 3 min timeout to avoid mutex holder crash
                    redis.expire(keyMutex, 3 * 60);
                    String dbValue = db.get(key);
                    redis.set(key, dbValue);
                    redis.delete(keyMutex);
                }
            }
        });
    }
    return value;
}

4. 资源保护(熔断技术)

利用 netflix 的 hystrix，可以做资源的隔离保护主线程池。

方案对比

解决方案	优点	缺点
简单分布式锁	1. 思路简单 2. 保证一致性	1. 代码复杂度增大 2. 存在死锁的风险 3. 存在线程池阻塞的风险
加另外一个过期时间	保证一致性	1. 代码复杂度增大 2. 存在死锁的风险 3. 存在线程池阻塞的风险
不过期	异步构建缓存，不会阻塞线程池	1. 不保证一致性 2. 代码复杂度增大(每个value都要维护一个timekey) 3. 占用一定的内存空间(每个value都要维护一个timekey)
资源隔离组件hystrix	1. hystrix技术成熟，有效保证后端 2. hystrix监控强大	1. 部分访问存在降级策略

总结

1. 热点key + 过期时间 + 复杂的构建缓存过程 => mutex key问题
2. 构建缓存一个线程做就可以了
3. 四种解决方案：没有最佳只有最合适

赠:

下面是JedisPool优化方法的文章:

• Jedis常见异常汇总
• JedisPool资源池优化

666 彩蛋

刚开始写博客, 希望大家支持, 如果有没疑问或不清楚的地方可以留言噢!

下周再见~