初识Redis

SET key value

设置指定 key 的值

GET key

获取指定 key 的值

GETRANGE kay start end

返回 key 中字符串值的子字符

GETSET key value

将给定 key 的值设为 value ，并返回 key 的旧值(old value)。

HSET key field value

将哈希表 key 中的字段 field 的值设为 value 。

HGET key field

获取存储在哈希表中指定字段的值。

HDELETE key field1[field2]

删除一个或多个哈希表字段

LPUSH key value1[value2]

将一个或多个值插入到列表头部

LPUSHX key value

将一个值插入到已存在的列表头部

LPOP key

移出并获取列表的第一个元素

LLEN key

获取列表长度

SADD key member1 [member2]

向集合添加一个或多个成员

SMEMBERS key

返回集合中的所有成员

SISMEMBER key member

判断 member 元素是否是集合 key 的成员

SPOP key

移除并返回集合中的一个随机元素

ZADD key score1 member1 [score2 member2]

向有序集合添加一个或多个成员，或者更新已存在成员的分数

ZCARD key

获取有序集合的成员数

ZCOUNT key min max

计算在有序集合中指定区间分数的成员数

SUBSCRIBE channel [channel ...]

订阅给定的一个或多个频道的信息。

PSUBSCRIBE pattern [pattern ...]

订阅一个或多个符合给定模式的频道。

UNSUBSCRIBE [channel [channel ...]]

退订给定的频道。

PUNSUBSCRIBE [pattern [pattern ...]]

退订所有给定模式的频道。

PUBLISH channel message

将信息发送到指定的频道。

PUBSUB subcommand [argument [argument ...]]

查看订阅与发布系统状态。

穿透

描述：

   指在redis中查询一个一定不存在的数据时，需要从数据库中查询。查询到数据不存在时则不写入缓存，这将导致每次在redis中获取不存在的数据时都会去数据库中查询，这将导致缓存穿透。

解决方法：

1. 使用布隆过滤，将所有可能存在的数据哈希到一个足够大的bitmap中，一个一定不存在的数据就会被拦截，这样就可以避免对数据库的查询压力。
2. 如果在数据库查询的数据为空，不管何种原因造成的，仍然将空结果进行缓存，只是设置的过期时间很短，最长不超过5分钟。

缓存雪崩

描述：

   Redis的缓存集中在一段时间内失效，发生大量的穿透，将所有的查询落到数据库中，造成了缓存雪崩。

解决方法：

   1、缓存失效后，通过加锁或队列来控制数据库写缓存的线程数量。如一个key只允许一个线程查询数据和写缓存，其他线程等待。

2、可以通过缓存reload机制，预先去更新缓存，再即将发生大并发访问前手动触发加载缓存。

3、不同的key设置不同的过期失效时间，让缓存失效时间点尽量均匀。

4、做好二级缓存或者双缓存策略。A1为原始缓存，A2为拷贝缓存，A1失效时，可以访问A2，A1缓存失效时间设置为短期，A2设置为长期。

public class RedisLock {
    private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    private final JedisPool jedisPool;

    public RedisLock(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    /**
     * 尝试获取分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁
     * @param requestId 请求标识
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {

        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);

        if (LOCK_SUCCESS.equals(result)) {
            return true;
        }
        return false;

    }

    /**
     * 加锁
     * @param locaName  锁的key
     * @param acquireTimeout  获取超时时间
     * @param timeout   锁的超时时间
     * @return 锁标识
     */
    public String lockWithTimeout(String locaName,
                                  long acquireTimeout, long timeout) {
        Jedis conn = null;
        String retIdentifier = null;
        try {
            // 获取连接
            conn = jedisPool.getResource();
            // 随机生成一个value
            String identifier = UUID.randomUUID().toString();
            // 锁名，即key值
            String lockKey = "lock:" + locaName;
            // 超时时间，上锁后超过此时间则自动释放锁
            int lockExpire = (int)(timeout / 1000);

            // 获取锁的超时时间，超过这个时间则放弃获取锁
            long end = System.currentTimeMillis() + acquireTimeout;
            while (System.currentTimeMillis() < end) {
                if (conn.setnx(lockKey, identifier) == 1) {
                    conn.expire(lockKey, lockExpire);
                    // 返回value值，用于释放锁时间确认
                    retIdentifier = identifier;
                    return retIdentifier;
                }
                // 返回-1代表key没有设置超时时间，为key设置一个超时时间
                if (conn.ttl(lockKey) == -1) {
                    conn.expire(lockKey, lockExpire);
                }

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retIdentifier;
    }

    /**
     * 释放锁
     * @param lockName 锁的key
     * @param identifier    释放锁的标识
     * @return
     */
    public boolean releaseLock(String lockName, String identifier) {
        Jedis conn = null;
        String lockKey = "lock:" + lockName;
        boolean retFlag = false;
        try {
            conn = jedisPool.getResource();
            while (true) {
                // 监视lock，准备开始事务
                conn.watch(lockKey);
                // 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁
                if (identifier.equals(conn.get(lockKey))) {
                    Transaction transaction = conn.multi();
                    transaction.del(lockKey);
                    List<Object> results = transaction.exec();
                    if (results == null) {
                        continue;
                    }
                    retFlag = true;
                }
                conn.unwatch();
                break;
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retFlag;
    }
}

watch stock

get stock $stock

$stock = $stock - 10

multi

set stock $stock

exec

WATCH key [key ...]

监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

UNWATCH

取消 WATCH 命令对所有 key 的监视。

MULTI

标记一个事务块的开始。

EXEC

执行所有事务块内的命令。

DISCARD

取消事务，放弃执行事务块内的所有命令。

save 900 1  #在900秒(15分钟)之后，如果至少有1个key发生变化，则dump内存快照。

save 300 10 #在300秒(5分钟)之后，如果至少有10个key发生变化，则dump内存快照。

save 60 10000 #在60秒(1分钟)之后，如果至少有10000个key发生变化，则dump内存快照。

appendfsync always  #每次有数据修改发生时都会写入AOF文件。

appendfsync everysec  #每秒钟同步一次，该策略为AOF的缺省策略。

appendfsync no     #从不同步。高效但是数据不会被持久化。

优点

缺点

1. 利用多台计算机内存的和值，允许我们构造更大的数据库。
2. 通过多核和多台计算机，允许我们扩展计算能力；通过多台计算机和网络适配器，允许我们扩展网络带宽。

1. 涉及多个key的操作通常是不被支持的。举例来说，当两个set映射到不同的redis实例上时，你就不能对这两个set执行交集操作。
2. 涉及多个key的redis事务不能使用。
3. 当使用分区时，数据处理较为复杂，比如你需要处理多个rdb/aof文件，并且从多个实例和主机备份持久化文件。
4. 增加或删除容量也比较复杂。redis集群大多数支持在运行时增加、删除节点的透明数据平衡的能力，但是类似于客户端分区、代理等其他系统则不支持这项特性。然而，一种叫做presharding的技术对此是有帮助的。