一、键值设计

1. 1 key名定义规则

• 【建议】: 可读性和可管理性与不可重复性

以业务名(或数据库名或微服务名)为前缀(防止key冲突)，用冒号分隔，比如业务名:表名（或子业务名称）:id

order:sku:1001

举例：上面定义是定单业务下面的sku的id作为缓存key

• 【建议】：简洁性

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

user:{uid}:friends:messages:{mid}简化为u:{uid}:fr:m:{mid}

• 【强制】：不要包含特殊字符

不要包含空格、换行、单双引号以及其他转义字符

1.2 value设计

1.2.1 【强制】：拒绝bigkey

为什么要拒绝bigkey: Redis实例集群主要思想是将redis数据的key进行散列，通过hash函数特定的key会映射到指定的redis节点上,bigkey当这个KEY流量较大的时候，流量全打到一个Redis节点上，big key既占用比较多的内存，也可能占用比较多的网卡资源，造成redis阻塞。

防止热点问题（数据倾斜）,Redis big key也就意味着每次获取要产生的网络流量较大，假设一个bigkey为1MB，客户端每秒访问量为1000，那么每秒产生1000MB的流量，对于普通的千兆网卡(按照字节算是128MB/s)的服务器来说简直是灭顶之灾。

String类型控制在10KB以内，hash、list、set、zset元素个数不要超过5000。

1.2.2 bigkey举例说明

举例1：我们要设计一个网关过滤IP黑名单的缓存，有两种设计方法：

反例，使用LIST或SET格式来存储IP：

LPUSH lgateway:ipblack 10.25.63.123
LPUSH lgateway:ipblack 10.25.63.124
LPUSH lgateway:ipblack 10.25.63.125
LPUSH lgateway:ipblack 10.25.63.126    
...................................    

正例，使用String来存储IP：

SET gateway:ipblack:10.25.63.123 1
SET gateway:ipblack:10.25.63.124 1
SET gateway:ipblack:10.25.63.125 1
SET gateway:ipblack:10.25.63.126 1    

相对第二种方法，方法一随着IP数据与流量的增加，会产生热点问题，所以选择方法二

举例2：用户表的缓存

用户表结构如下：

CREATE TABLE `users` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `name` varchar(100) DEFAULT NULL,
  `password` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8

反例，使用map来存：

HGET 来存储 key为“userinfo”,value使用map存所有的用户信息

HGET userinfo 12321 {"name":"peter","age":22}
HGET userinfo 12322 {"name":"jake","age":23}
HGET userinfo 12323 {"name":"rose","age":24}
HGET userinfo 12324 {"name":"xiaomi","age":25}    
...................................    

正例，使用map来存储：

拆分行纪录来存储，key为 user: id, value为map

HMSET user:12321 name peter age 22
HMSET user:12322 name jake age 23
HMSET user:12323 name rose age 24
HMSET user:12324 name xiaomi age 25   
..................................

相对第二种方法，方法一随着用户数据与流量的增加，会产生热点问题，所以选择方法二。

1.2.3【推荐】：选择适合的数据类型。

例如：实体类型(要合理控制和使用数据结构，但也要注意节省内存和性能之间的平衡)

比如对用户信息的缓存，如下两例所示：

反例：

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

正例:

hmset user:1 name tom age 19 favor football

第一种方式得查多次才能得到这条纪录的值，第二种Map结构可以很方便的一次获取所有字段的值或获取某个字段的值。

1.3【强制】：控制key的生命周期

不是需要长期保存的数据得使用expire设置过期时间，临时缓存数据不允许长期保存，成为垃圾数据。比如短信验证码，有效期是3分钟，那行设置三分钟后过期，如下所示：

SET user:sms:userid 123456
EXPIRE user:sms:userid 180

二、命令使用

2.1【推荐】 O(N)命令关注N的数量

例如hgetall、lrange、smembers、zrange、sinter等并非不能使用，但是需要明确N的值（要限制每次取的数量）。有遍历的需求可以使用hscan、sscan、zscan代替。

2.2【推荐】：禁用命令

禁止线上使用keys、flushall、flushdb等，或者使用scan的方式渐进式处理。

警告：KEYS命令不能用在生产的环境中，这个时候如果数量过大效率是十分低的。同时也不要用KEYS正则匹配，官方建议直接用集合类型。

FLUSHALL，FLUSHDB 命令会清空数据，而且从不失败，对于线上集群非常危险。

三、【推荐】Redisson实现分布式锁

使用Redis分布式锁开发时推荐使用Redisson客户端。

3.1 Reddsion介绍

官方推荐的Redisson实现Redis分布式锁，Redisson功能强大，几行代码就可以实现Jedis客户端中需要实现的复杂锁的功能。

Reddisson可以防止死锁：如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

Redisson提供了很多种类型的分布式锁和分布式同步器，如下：

1. 可重入锁（Reentrant Lock） 
2. 公平锁（Fair Lock）       
3. 联锁（MultiLock）
4. 红锁（RedLock）
5. 读写锁（ReadWriteLock）
6. 信号量（Semaphore）
7. 可过期性信号量（PermitExpirableSemaphore）
8. 闭锁（CountDownLatch）

3.2 Springboot结合Redission使用

3.2.1 pom依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.9.0</version>
</dependency>

3.2.2 application.yml

##端口号
server:
  port: 8080
#redis
redis:
  address:
    #单机模式 
    master: redis://127.0.0.1:6379
    #主从模式
    slave: redis://127.0.0.1:6379
    #集群模式
    cluster: redis://127.0.0.1:7001,redis://127.0.0.1:7002,redis://127.0.0.1:7003,redis://127.0.0.1:7004,redis://127.0.0.1:7005,redis://127.0.0.1:7006
  #连接模式 1 单机模式 2 主从模式 3 集群模式
  connectionMode: 1

3.2.3 Redisson配置类

支持三种配置模式：单机模式、主从模式、集群模式

@Configuration
public class RedissionConfig {
    private static final Logger logger = LoggerFactory.getLogger(RedissionConfig.class);

    @Autowired
    private Environment env;
    private Config config;

    private static final String REDISSION_ClIENT_NAME = "SHIT_IS_PERFECT";

    private static final Integer POOL_SIZE = 200;

    private static final Integer IDLE_SIZE = 15;

    private static final Integer TIME_OUT = 10000;

    @Value("${redis.address.master}")
    private String master;
    @Value("${redis.address.slave}")
    private String slave;
    @Value("${redis.connectionMode}")
    private Integer connectionMode;
    @Value("${redis.address.cluster}")
    private String cluster;

    @Bean
    public RedissonClient redissionClient(){
        config = new Config();
        switch (connectionMode){
            //standalone 单机模式
            case 1:
                config.useSingleServer().setAddress(master)
                        .setConnectionPoolSize(POOL_SIZE)
                        .setClientName(REDISSION_ClIENT_NAME)
                        .setConnectionMinimumIdleSize(IDLE_SIZE)
                        .setConnectTimeout(TIME_OUT)
                        .setPingTimeout(30000);
                break;
             //master_slave 主从模式
            case 2:
                config.useMasterSlaveServers().setClientName(REDISSION_ClIENT_NAME)
                        .setConnectTimeout(TIME_OUT)
                        .setMasterAddress(master)
                        .addSlaveAddress(slave)
                        .setMasterConnectionMinimumIdleSize(IDLE_SIZE)
                        .setMasterConnectionPoolSize(POOL_SIZE)
                        .setSlaveConnectionMinimumIdleSize(IDLE_SIZE)
                        .setSlaveConnectionPoolSize(POOL_SIZE);
                break;
             //cluster 集群模式
            case 3:
                config.useClusterServers()
                        .addNodeAddress(cluster.split(","))//这是用的集群server
                        .setScanInterval(2000) //设置集群状态扫描时间
                        .setMasterConnectionPoolSize(10000) //设置连接数
                        .setSlaveConnectionPoolSize(10000);
                break;
        }
        return Redisson.create(config);
    }
}

3.2.4 RedissonLocker的具体实现类

@Service("redissonLocker")
public class RedissonLockerImpl implements DistributedLocker {

    @Autowired
    private RedissonClient redissonClient;

    /**************************可重入锁**************************/
    /**
     * 拿不到lock就不罢休，不然线程就一直block（阻塞）
     * 没有超时时间,默认30s
     *
     * @param lockKey
     * @return
     */
    @Override
    public RLock lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock();
        return lock;
    }

    /**
     * 自己设置超时时间
     *
     * @param lockKey 锁的key
     * @param timeout 秒 如果是-1，直到自己解锁，否则不会自动解锁
     * @return
     */
    @Override
    public RLock lock(String lockKey, int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, TimeUnit.SECONDS);
        return lock;
    }

    /**
     * 自己设置超时时间
     *
     * @param lockKey 锁的key
     * @param unit 锁时间单位
     * @param timeout 超时时间
     *
     */
    @Override
    public RLock lock(String lockKey, TimeUnit unit, int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, unit);
        return lock;
    }

    /**
     *  尝试加锁，最多等待waitTime，上锁以后leaseTime自动解锁
     * @param lockKey   锁key
     * @param unit      锁时间单位
     * @param waitTime  等到最大时间
     * @param leaseTime 锁失效时间
     * @return 如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false
     */
    @Override
    public boolean tryLock(String lockKey, TimeUnit unit, int waitTime, int leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }



    /**************************公平锁**************************/
    /**
     *   尝试加锁，最多等待waitTime，上锁以后leaseTime自动解锁
     * @param lockKey   锁key
     * @param unit      锁时间单位
     * @param waitTime  等到最大时间
     * @param leaseTime 锁失效时间
     * @return 如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false
     */
    public boolean fairLock(String lockKey, TimeUnit unit, int waitTime, int leaseTime) {
        RLock fairLock = redissonClient.getFairLock(lockKey);
        try {
            return fairLock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 释放锁
     * @param lockKey 锁key
     */
    @Override
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.unlock();
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock(RLock lock) {
        lock.unlock();
    }
}

四、附录：删除bigkey

4.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);
}

4.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);
}

4.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.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);
}