Redis_随笔_内存溢出

Redis

1. NoSQL
2. Redis简介
3. Redis的五大数据类型
4. Redis的三大特殊数据类型
5. Redis中的事务
- 5.1 监控（Watch，常被当做乐观锁使用）
- 5.2 悲观锁
6. Redis发布订阅
7. Redis哨兵模式
8. Redis缓存穿透、击穿、雪崩
- 8.1 缓存穿透（频繁查询不存在的数据）
- 8.2 缓存击穿（频繁查询热点key）
- 8.3 缓存雪崩（大量的缓存集体失效）
9. SpringBoot集成Redis
10. Redis官网

1. NoSQL

在过去很长的一段时间里，如MySQL、Oracel等结构化查询语言（Structured Query Language, SQL）数据库一直是数据存储的主要方式，它们的结构类似一张Excel表，建表时就明确规定了每一列的属性。这种“严格”的数据库表形式在互联网发展的初期适用，但随着近些年用户数据爆炸式的增长，传统的SQL型数据库已经很难完全满足我们的需求，在这种情况下，NoSQL就展现出了它的优势。

NoSQL，全称是Not only SQL，也常被称作非关系型的数据库。NoSQL放弃了传统关系型数据库中数据之间的关联性，便于扩展，同时拥有非常高的读写性能。例如，传统的关系型数据库中，往数据量较大的数据库中增加一列的代价是十分巨大的，而NoSQL就能很好的解决这一问题。总的来说，NoSQL可以分为以下几类：

键值(Key-Value)存储数据库：以键值对形式存储的非关系型数据库，读写效率高，查询快，常被用于缓存。
列存储数据库：通常是用来应对分布式存储的海量数据。
文档型数据库：与键值存储相类似，但其允许嵌套键值，在处理网页等复杂数据时，文档型数据库比传统键值数据库的查询效率更高。
图形(Graph)数据库：专注于构建关系图谱，社交网络，推荐系统等。

2. Redis简介

Redis是应用最为广泛的K-V存储数据库，最常见的用途就是作为缓存，除此之外还可以作为数据库和消息中间件。Redis基于内存进行 *** 作，且基于单线程实现，至于为什么是单线程，简单的来说就是“够用了”。

单从效率上来讲CPU > 内存 > 磁盘，因此Redis性能的瓶颈不是CPU，而是内存和网络带宽。虽然我们常说“多线程的速度比较快“，但事实上单线程并不一定会比多线程慢，这取决于具体的业务场景。对于Redis而言，单线程完全可以满足其需求；由于线程间切换也会消耗一部分的时间，使用多线程实现的话，如果线程间的切换比较频繁，反而会在不必要的地方消耗大量的时间，降低Redis的性能。

安装完Redis之后会有一个默认的redis.conf配置文件，通过配置文件可以对Redis进行一些设置，主要包括：

网络：绑定的ip、是否开启保护模式、端口设置等
通用：是否以守护进程的方式运行、日志目录、数据库数量等
持久化：RDB、AOF
主从复制
安全
客户端：客户端最大可连接数、redis最大内存容量、达到最大内存容量后的处理策略等

默认的：

Redis默认有16个数据库（0 ~ 15）；
默认使用的是第0个数据库（可以使用select进行切换）；
默认使用RDB进行持久化

3. Redis的五大数据类型

String类型：

最常用的类型；
具备自增和自减 *** 作，可作为计数器；
可用于对象的缓存；
setnx：如果不存在则新建，存在则不执行；在分布式锁中常常会用到；

List类型（以L开头的命令）：

列表，可用于缓存一组数据集合；
可作为栈（Stack）；
可作为队列（Queue）。

Set类型（以S开头的命令）：

与Java中的Set类似，它保证了数据的唯一性，但不保证数据的有序性。

Hash（以H开头的命令）：

与Java的HashMap类似；
可用于存储经常变更的对象，如用户信息之类的数据。

Zset（有序集合，以Z开头的命令）：

本质上也是Set，只是在其基础上增加了一个用于排序的score。

4. Redis的三大特殊数据类型

Geospatial（地理位置，以Geo开头的命令）：

基于Zset实现，可以用Zset的命令 *** 作它；
以经纬度为基础，可用于定位；
可以进行一些简单的位置计算：如获取两地点间的距离、在某个范围内进行搜索等。

HyperLogLog（基数，以Pf开头的命令）：

基数：简单的来说就是不重复的数，{1, 3}和{2, 3}的基数为3（{1, 2, 3}），可以接受误差；
最大的优点是所占用的内存少；
应用场景（可接受误差的统计场景可以使用）：如独立访问量（UV）。

Bitmaps（位图）：

Bitmaps本身不是一种数据结构，它实际上就是字符串，但是它可以对字符串进行位 *** 作。
使用场景：登陆/未登录、活跃/不活跃、打卡/未打卡等。

5. Redis中的事务

一般来说，我们认为Redis单条命令是原子的，但其事务不保证原子性。

Redis事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。EXEC 命令触发事务中所有命令的执行，因此，如果客户机在调用 EXEC 命令之前丢失了与事务上下文中的服务器的连接，则不执行任何 *** 作；如果调用 EXEC 命令，则执行所有 *** 作。

看到这里也许会有读者产生疑问了，这不就保证了Redis的原子性了吗。需要注意的是，如果在 EXEC 之后发生了错误，Redis不会以特殊的方式处理：也就是说，即使在事务过程中某些命令失败，所有其它命令也将被执行。并且Redis不支持回滚，如果 Redis 服务器崩溃或者被系统管理员服务器以某种困难的方式杀死，那么这时候可能只有部分 *** 作被执行了。针对此，Redis的作者是这样解释的：

首先，MySQL 和 Redis 的定位不一样，一个是关系型数据库，一个是 NoSQL。
MySQL 的 SQL 查询是可以相当复杂的，而且 MySQL 没有事务队列这种说法，SQL 真正开始执行才会进行分析和检查，MySQL 不可能提前知道下一条 SQL 是否正确，所以支持事务回滚是非常有必要的。
但是，Redis 使用了事务队列来预先将执行命令存储起来，并且会对其进行格式检查的，提前就知道命令是否可执行了。所以如果只要有一个命令是错误的，那么这个事务是不能执行的。
Redis 作者认为基本只会出现在开发环境的编程错误其实在生产环境基本是不可能出现的（例如对 String 类型的数据库键执行 LPUSH *** 作），所以他觉得没必要为了这事务回滚机制而改变 Redis 追求简单高效的设计主旨。

简单的来说，关于Redis是否具有原子性：

出现语法性错误，将不能通过检查，事物将不会被执行；
出现非语法性错误（如类型错误等），事物会被执行，除了有错误的语句，其他正确的语句会被正常执行。

Redis事务的流程：

开启事务（multi）；
命令入队（写入要执行的命令）；
执行事物（exec），可以用discard命令取消事物；
特点：一次性、顺序性、排他性。

5.1 监控（Watch，常被当做乐观锁使用）

Redis中的Watch可以给事物加锁。

# 相当于乐观锁，CAS *** 作
127.0.0.1:6379> set money 100
OK
127.0.0.1:6379> set out 0
OK
127.0.0.1:6379> watch money # 开始监视 money
OK
127.0.0.1:6379> multi #开启事务
OK
127.0.0.1:6379> decrby money 10
QUEUED
127.0.0.1:6379> incrby out 10
QUEUED
127.0.0.1:6379> exec # 执行事务（执行前会先判断监视的key是否被修改过，如果被修改了则事务执行失败，返回nil）
1) (integer) 90
2) (integer) 10
# 事务执行成功之后，之前设置的watch自动失效

unwatch # 如果事务执行失败，则可以通过unwatch命令取消监视后，再重新开启监视并执行事务

5.2 悲观锁

setnx常被用于分布式悲观锁中，如果在公司里落地生产环境用分布式锁的时候，一般会用开源类库的，比如Redis分布式锁，一般就是用Redisson框架就好了，简便易用。

RLock lock = redisson.getLock("myLock");
lock.lock(); // 加锁
lock.unlock() // 释放锁

简单加锁的过程：

判断redis中是否存在key：myLock；
如果存在，则表示此锁已被其他人持有，进入循环继续等待并申请；
如果不存在，则持有锁：在Redis中添加key为myLock的字段，value的值为客户端ID，一般会设置自动过期时间；
如果达到自动过期时间后，该客户端依旧还持有改锁，则自动延长key的生存时间；
可以使用Hash结构存储value：客户端ID – 加锁次数，实现可重入锁的机制；

存在的问题：

在实际生产环境中，一般会使用哨兵模式（master & slave）保证数据的安全性；
当我们在master中写入myLock后，该值也会被异步复制到所有的slave上；
如果这个过程中，master宕机了，发生了主从切换，就可能会产生问题：
- A先在master上获取了锁，然后master宕机了，myLock还未被复制到slave上；
- slave被切换为主机，此时B也申请myLock，在新的主机上持有了改锁；
- 此时，A和B都会认为自己持有了锁myLock，会产生数据一致性问题。

6. Redis发布订阅

Redis 发布订阅（pub/sub）是一种消息通讯模式：

发送者（pub）推送消息；
订阅者（sub）接收消息；
Redis客户端可以订阅任意数量的频道。

Redis可以通过PUBLISH、SUBSCRIBE和PSUBSCRIBE等命令实现发布订阅的功能：

# 订阅端
127.0.0.1:6379> SUBSCRIBE jublog # 订阅一个频道
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "jublog"
3) (integer) 1
# 开始等待接收推送的消息

# 发布端向jublog推送消息后，订阅段会接收到对应的消息
1) "message"
2) "jublog"
3) "hello,Redis"

# 发布端
127.0.0.1:6379> PUBLISH jublog "hello,Redis" # 向指定频道推送消息
(integer) 1

基本原理：

redis-server 里维护了一个字典（key代表了一个单独的channel，value是一个链表，保存了所有订阅了这个channel的客户端）；
当客户端通过SUBSCRIBE命令订阅了某个channel后，该客户端就会被添加到该channel对应的链表中去；
PUBLISH命令首先会获取指定的channel所指向的链表，然后遍历并将消息发送给每一个订阅了的客户端。

7. Redis哨兵模式

主从切换：当主服务器宕机后，需要手动把一台从服务器切换为主服务器，需要人力的干预，且在这一段时间内服务是不可用的，Redis从2.8开始正式提供了Sentinel（哨兵）架构来解决这一问题。

哨兵模式可以理解为一种自动版的主从切换模式：

Redis中哨兵是一个独立的进程；
哨兵可以通过发送命令，并等待Redis服务器响应，以监控多个Redis实例；
当哨兵监测到master（主服务器）宕机后，会通过投票机制自动的将一台slave（从服务器）切换为主服务器，并通过发布订阅模式通知其他的从服务器重新“跟随”新的master；
可以设置多个哨兵，并让哨兵之间相互监控，构成多哨兵模式，以避免单个哨兵出现问题后系统无法正常运行的情况。

简单的单哨兵示意图如下所示：

8. Redis缓存穿透、击穿、雪崩 8.1 缓存穿透（频繁查询不存在的数据）

简单的来说，缓存穿透就是频繁的查询不存在的数据：

查询一个数据，发现Redis缓存中不存在，于是就向持久层的数据库查询；
由于数据不存在，持久层中的查询结果当然也是null，本次查询失败；
当大量类似的请求出现时，由于缓存均未命中，它们都会向持久层的数据库进行查询，出现缓存穿透问题（大量的无效查询给持久层数据库造成了很大的压力）。

解决方法主要有缓存空对象和布隆过滤器：

缓存空对象很好理解，就是当持久层的查询结果为空对象时，也将其存储下来，那么下次再次查询的时候就不用再访问持久层的数据库了。当然，这样做也会存在一些问题，存储空值不可避免的会造成储存空间的浪费，即使设置了过期时间，也可能会因为缓存和持久层在某个时间区间内数据不一致，而对一些需要保持一致性的业务造成影响。
布隆过滤器是一种特殊的数据结构，他把所有可能的查询参数以Hash的形式进行存储，在控制层先进行一次校验；符合则继续，不符合则丢弃，从而减轻了持久层数据库的查询压力。

8.2 缓存击穿（频繁查询热点key）

缓存击穿是指Redis中存在一个十分热点key，这个key每一时刻都承受着巨大的并发量，当这个key达到过期时间失效的瞬间，大并发的查询就穿透的缓存，直接请求持久层的数据库，导致持久层数据库瞬间压力过大。

解决方案：

设置热点数据的缓存永不过期；
互斥锁：使用分布式锁，保证对于每个key同时只能有一个线程查询后端的服务，其他没有获得权限的线程则继续等待；相当于是把高并发的压力转移到了分布式锁上，对分布式锁的压力较大。

8.3 缓存雪崩（大量的缓存集体失效）

缓存雪崩是指在某一个时间段，大量的缓存集体失效。例如双十一的时候，会将热点商品先一步缓存到Redis中，并设置了它们的缓存时间。由于这些商品几乎是在同一时间被缓存进Redis中的，如果它们的过期时间也相同的话，就会在某一个时刻集体失效，此时针对这些商品的查询都落到了持久层的数据库上，从而产生周期性的压力波峰。当然，最坏的情况就是Redis服务器宕机，所有缓存直接全部失效，所有查询都直接访问持久层的数据库，很容易让持久层的服务器直接挂掉。

解决方案：

Redis高可用：搭建Redis集群，多增设几台Redis服务器，即使一台挂掉了其他的也能够继续工作；
限流降级：缓存失效后，通过加锁或者队列控制访问数据库的线程数量；
数据预热：先将热点数据缓存进Redis中，并设置不同的过期时间，使失效时间分布的尽量均匀。

9. SpringBoot集成Redis

SpringBoot集成Redis需要引入对应的SpringData依赖：


    org.springframework.boot
    spring-boot-starter-data-redis

配置文件中增加相关配置（yml）：

spring:
  redis:
    host: 127.0.0.1 # 默认ip
    port: 6379 # 默认端口
    password: your password

解决Redis乱码问题：

@Configuration
public class RedisConfig {

    // 自己定义的RedisTemplate
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate redisTemplate(RedisConnectionFactory factory) {
        // 为方便日常使用，一般定义为类型
        RedisTemplate template = new RedisTemplate();
        template.setConnectionFactory(factory);
        
        // Json序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // String序列化配置
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value序列化方式采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value序列化方式采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet(); // 应用此模板

        return template;
    }
}

为了方便使用，一般会定义自己的工具类，下面给出了一个参考：

@Component
public final class RedisUtil {

    private RedisTemplate redisTemplate;

    public RedisTemplate getRedisTemplate() {
        return redisTemplate;
    }

    @Autowired
    public void setRedisTemplate(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    
    public boolean expire(String key, long time) {
        try {
            if (time > 0) {
                redisTemplate.expire(key, time, TimeUnit.SECONDS);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    
    public long getExpire(String key) {
        return redisTemplate.getExpire(key, TimeUnit.SECONDS);
    }


    
    public boolean hasKey(String key) {
        try {
            return redisTemplate.hasKey(key);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    
    @SuppressWarnings("unchecked")
    public void del(String... key) {
        if (key != null && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            } else {
                redisTemplate.delete((Collection) CollectionUtils.arrayToList(key));
            }
        }
    }


    // ============================String=============================

    
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }

    

    public boolean set(String key, Object value) {
        try {
            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    

    public boolean set(String key, Object value, long time) {
        try {
            if (time > 0) {
                redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
            } else {
                set(key, value);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }


    
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }


    // ================================Map=================================

    
    public Object hget(String key, String item) {
        return redisTemplate.opsForHash().get(key, item);
    }

    
    public Map hmget(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    
    public boolean hmset(String key, Map map) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    
    public boolean hmset(String key, Map map, long time) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    
    public boolean hset(String key, String item, Object value) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    
    public boolean hset(String key, String item, Object value, long time) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    
    public void hdel(String key, Object... item) {
        redisTemplate.opsForHash().delete(key, item);
    }


    
    public boolean hHasKey(String key, String item) {
        return redisTemplate.opsForHash().hasKey(key, item);
    }


    
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }


    
    public double hdecr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, -by);
    }


    // ============================set=============================

    
    public Set

Redis

发表评论

评论列表（0条）