Redis 一篇搞懂，就硬懂

Redis是什么?

redis是一种非关系性数据库，它是由c语言开发的一个开源高性能的键值对内存数据库，可以用作于缓存，数据库，消息中间件等。

为什么redis这么快，有哪些用处？

redis数据主要在内存中读写，支持10w的并发量
单线程，线程安全，采用了IO多路复用机制
丰富的数据类型，String，list，set，zset，hash
支持数据的持久化，可以将内存保存在磁盘中，有自己的持久化策略
主从复制，哨兵模式，高可用
可以用作分布式锁
可以作为消息中间件使用，支持发布订阅

五种数据类型和使用场景

1、string是redis最基本的类型，可以理解成与memcached一模一样的类型，一个key对应一个value。

value不仅是string，也可以是数字。string类型是二进制安全的，意思是redis的string类型可以包含任何数据，比如jpg图片或者序列化的对象。string类型的值最大能存储512M。

2、Hash是一个键值（key-value）的集合。

redis的hash是一个string的key和value的映射表，Hash特别适合存储对象。常用命令：hget,hset,hgetall等。

3、list列表是简单的字符串列表，按照插入顺序排序。

可以添加一个元素到列表的头部（左边）或者尾部（右边）常用命令：lpush、rpush、lpop、rpop、lrange(获取列表片段)等。

应用场景：list应用场景非常多，也是Redis最重要的数据结构之一，比如twitter的关注列表，粉丝列表都可以用list结构来实现。

数据结构：list就是链表，可以用来当消息队列用。redis提供了List的push和pop操作，还提供了操作某一段的api，可以直接查询或者删除某一段的元素。

实现方式：redis list的是实现是一个双向链表，既可以支持反向查找和遍历，更方便操作，不过带来了额外的内存开销。

4、set是string类型的无序集合。

集合是通过hashtable实现的。set中的元素是没有顺序的，而且是没有重复的。

常用命令：sdd、spop、smembers、sunion等。

应用场景：redis set对外提供的功能和list一样是一个列表，特殊之处在于set是自动去重的，而且set提供了判断某个成员是否在一个set集合中。

5、zset和set一样是string类型元素的集合，且不允许重复的元素。

常用命令：zadd、zrange、zrem、zcard等。

使用场景：sorted set可以通过用户额外提供一个优先级（score）的参数来为成员排序，并且是插入有序的，即自动排序。

当你需要一个有序的并且不重复的集合列表，那么可以选择sorted set结构。

和set相比，sorted set关联了一个double类型权重的参数score，使得集合中的元素能够按照score进行有序排列，redis正是通过分数来为集合中的成员进行从小到大的排序。

实现方式：Redis sorted set的内部使用HashMap和跳跃表(skipList)来保证数据的存储和有序，HashMap里放的是成员到score的映射，而跳跃表里存放的是所有的成员，排序依据是HashMap里存的score

使用跳跃表的结构可以获得比较高的查找效率，并且在实现上比较简单。

redis具体怎么用的？

一般有两种方式：一种是通过redistemplate调用方法来使用的，另一种方法是用spring的注解使用的。

redis在项目中的使用

在pom.xml文件下的依赖：

<!--redis整合-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.session</groupId>
            <artifactId>spring-session-data-redis</artifactId>
        </dependency>

spring-boot-starter-data-redis:在spring boot 2.x以后底层不再使用Jedis，而是换成了Lettuce。

commons-pool2：用作redis连接池，如不引入启动会报错

spring-session-data-redis：spring session引入，用作共享session。配置文件application.yml的配置：

redis:
  port: 6379
  password: root
  host: 127.0.0.1
  lettuce:
    pool:
      #redis缓存池
      max-active: 1000

创建实体类SysUser.java

@Data
@AllArgsConstructor
@NoArgsConstructor
public class SysUser implements Serializable {

    private static final long serialVersionUID = -1887540228012635046L;

    private Integer userId;

    private String userIp;

    private String userName;

    private String userPassword;

    private String userEmail;

    private String userProfilePhoto;

    private Date userRegistrationTime;

    private Date userBirthday;

    private Integer userAge;

    private String userTelephoneNumber;

    private String userNickname;

    private String userPer;

    private Boolean userStatus;
}

Redis配置

修改默认RedisTemplate，key和value都需要反序列化才能再redis的客户端中得到序列化之前的值，就无需在程序中再反序列话一次，所以自定义的模版是很必要的。且在配置类中加入@EnableCaching注解开启缓存，@Configuration注解将配置的bean注入到spring容器中

添加RedisConfig.java

@Configuration
@EnableCaching
public class RedisConfig {
    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory)
    {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);

        FastJson2JsonRedisSerializer serializer = new FastJson2JsonRedisSerializer(Object.class);

        ObjectMapper mapper = new ObjectMapper();
        mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        serializer.setObjectMapper(mapper);

        template.setValueSerializer(serializer);
        // 使用StringRedisSerializer来序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }
}

其中的使用了自定义的序列化方式FastJson2JsonRedisSerializer:

public class FastJson2JsonRedisSerializer<T> implements RedisSerializer<T>
{
    @SuppressWarnings("unused")
    private ObjectMapper objectMapper = new ObjectMapper();

    public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");

    private Class<T> clazz;

    static
    {
        ParserConfig.getGlobalInstance().setAutoTypeSupport(true);
    }

    public FastJson2JsonRedisSerializer(Class<T> clazz)
    {
        super();
        this.clazz = clazz;
    }

    public byte[] serialize(T t) throws SerializationException
    {
        if (t == null)
        {
            return new byte[0];
        }
        return JSON.toJSONString(t, SerializerFeature.WriteClassName).getBytes(DEFAULT_CHARSET);
    }

    public T deserialize(byte[] bytes) throws SerializationException
    {
        if (bytes == null || bytes.length <= 0)
        {
            return null;
        }
        String str = new String(bytes, DEFAULT_CHARSET);

        return JSON.parseObject(str, clazz);
    }

    public void setObjectMapper(ObjectMapper objectMapper)
    {
        Assert.notNull(objectMapper, "'objectMapper' must not be null");
        this.objectMapper = objectMapper;
    }

    protected JavaType getJavaType(Class<?> clazz)
    {
        return TypeFactory.defaultInstance().constructType(clazz);
    }
}

封装好的redis工具类进行缓存的各种操作。

创建RedisCache.java

@Component
public class RedisCache
{
    @Autowired
    public RedisTemplate redisTemplate;

    /**
     * 缓存基本的对象，Integer、String、实体类等
     * @param key 缓存的键值
     * @param value 缓存的值
     * @return 缓存的对象
     */
    public <T> ValueOperations<String, T> setCacheObject(String key, T value)
    {
        ValueOperations<String, T> operation = redisTemplate.opsForValue();
        operation.set(key, value);
        return operation;
    }

    /**
     * 缓存基本的对象，Integer、String、实体类等
     *
     * @param key 缓存的键值
     * @param value 缓存的值
     * @param timeout 时间
     * @param timeUnit 时间颗粒度
     * @return 缓存的对象
     */
    public <T> ValueOperations<String, T> setCacheObject(String key, T value, Integer timeout, TimeUnit timeUnit)
    {
        ValueOperations<String, T> operation = redisTemplate.opsForValue();
        operation.set(key, value, timeout, timeUnit);
        return operation;
    }

    /**
     * 获得缓存的基本对象。
     *
     * @param key 缓存键值
     * @return 缓存键值对应的数据
     */
    public <T> T getCacheObject(String key)
    {
        ValueOperations<String, T> operation = redisTemplate.opsForValue();
        return operation.get(key);
    }

    /**
     * 删除单个对象
     *
     * @param key
     */
    public void deleteObject(String key)
    {
        redisTemplate.delete(key);
    }

    /**
     * 删除集合对象
     *
     * @param collection
     */
    public void deleteObject(Collection collection)
    {
        redisTemplate.delete(collection);
    }

    /**
     * 缓存List数据
     *
     * @param key 缓存的键值
     * @param dataList 待缓存的List数据
     * @return 缓存的对象
     */
    public <T> ListOperations<String, T> setCacheList(String key, List<T> dataList)
    {
        ListOperations listOperation = redisTemplate.opsForList();
        if (null != dataList)
        {
            int size = dataList.size();
            for (int i = 0; i < size; i++)
            {
                listOperation.leftPush(key, dataList.get(i));
            }
        }
        return listOperation;
    }

    /**
     * 获得缓存的list对象
     *
     * @param key 缓存的键值
     * @return 缓存键值对应的数据
     */
    public <T> List<T> getCacheList(String key)
    {
        List<T> dataList = new ArrayList<T>();
        ListOperations<String, T> listOperation = redisTemplate.opsForList();
        Long size = listOperation.size(key);

        for (int i = 0; i < size; i++)
        {
            dataList.add(listOperation.index(key, i));
        }
        return dataList;
    }

    /**
     * 缓存Set
     *
     * @param key 缓存键值
     * @param dataSet 缓存的数据
     * @return 缓存数据的对象
     */
    public <T> BoundSetOperations<String, T> setCacheSet(String key, Set<T> dataSet)
    {
        BoundSetOperations<String, T> setOperation = redisTemplate.boundSetOps(key);
        Iterator<T> it = dataSet.iterator();
        while (it.hasNext())
        {
            setOperation.add(it.next());
        }
        return setOperation;
    }

    /**
     * 获得缓存的set
     *
     * @param key
     * @return
     */
    public <T> Set<T> getCacheSet(String key)
    {
        Set<T> dataSet = new HashSet<T>();
        BoundSetOperations<String, T> operation = redisTemplate.boundSetOps(key);
        dataSet = operation.members();
        return dataSet;
    }

    /**
     * 缓存Map
     *
     * @param key
     * @param dataMap
     * @return
     */
    public <T> HashOperations<String, String, T> setCacheMap(String key, Map<String, T> dataMap)
    {
        HashOperations hashOperations = redisTemplate.opsForHash();
        if (null != dataMap)
        {
            for (Map.Entry<String, T> entry : dataMap.entrySet())
            {
                hashOperations.put(key, entry.getKey(), entry.getValue());
            }
        }
        return hashOperations;
    }

    /**
     * 获得缓存的Map
     *
     * @param key
     * @return
     */
    public <T> Map<String, T> getCacheMap(String key)
    {
        Map<String, T> map = redisTemplate.opsForHash().entries(key);
        return map;
    }

    /**
     * 获得缓存的基本对象列表
     * 
     * @param pattern 字符串前缀
     * @return 对象列表
     */
    public Collection<String> keys(String pattern)
    {
        return redisTemplate.keys(pattern);
    }
}

定义接口，并且实现接口

@Service
public class SysUserServiceImpl implements SysUserService {

    @Resource
    private SysUserMapper userMapper;

    @Resource
    private RedisCache redisCache;

    public static Logger logger = Logger.getLogger(String.valueOf(SysUserServiceImpl.class));
    
    @Override
    public void save(SysUser user) {
        redisCache.setCacheObject(String.valueOf(user.getUserId()),user);
        logger.info("进入save方法，当前存储对象：｛｝"+user.toString());
    }

    @Override
    public void delete(int uid) {
        redisCache.deleteObject(String.valueOf(uid));
    }

    /*
     * @Description //先查redis里是否存在用户信息，若没有则从数据库中查询
     **/
    @Override
    public SysUser get(int uid) {
        SysUser user = redisCache.getCacheObject(String.valueOf(uid));
        if (user!=null){
            logger.info("进入get方法，当前对象：｛｝"+user.toString());
            return user;
        }else {
            logger.info("redis库里没有这条用户数据，需要通过数据库获取");
            SysUser sysUser = userMapper.selectByPrimaryKey((long) uid);
            save(sysUser);
            return sysUser;
        }
    }
}

这里是从数据库拿取数据，若缓存库没有，就去数据库查询数据并保存到缓存中。

在控制层进行测试：

@Controller
@RequestMapping("/user")
public class UserController {

    @Resource
    private SysUserService userService;

    @ResponseBody
    @GetMapping("/query/{uid}")
    public SysUser getUser(@PathVariable("uid") Integer uid){
        SysUser user = userService.get(uid);
        return user;
    }
}

启动springboot项目，并调用接口http://localhost:8080/user/query/1，查询id=1的用户。

缓存的注解使用

1、@Cacheable

根据方法的请求参数对其结果进行缓存

key：缓存的key，可以为空，如果指定要按照SPEL表达式编写，如果不指定，则按照方法的所有参数进行组合。
value：缓存的名称，必须指定至少一个（如 @Cacheable (value=’user’)或者@Cacheable(value={‘user1’,’user2’})）
condition：缓存的条件，可以为空，使用SPEL编写，返回true或者false，只有为true才进行缓存。

2、@CachePut

根据方法的请求参数对其结果进行缓存，和@Cacheable不同的是，它每次都会触发真实方法的调用。参数描述见上。

3、@CacheEvict

根据条件对缓存进行清空

key：同上
value：同上
condition：同上
allEntries：是否清空所有缓存内容，缺省为false，如果指定为true，则方法调用后将立即清空所有缓存
beforeInvocation：是否在方法执行前就清空，缺省为false，如果指定为true，则在方法还没有执行的时候就清空缓存。缺省情况下，如果方法执行抛出异常，则不会清空缓存。

缓存出现的问题

实际项目中缓存会遇到什么问题，而你是怎么解决的呢？

缓存穿透

缓存穿透的意思是当一些热点数据请求进缓存的时候，可能缓存的key已经失效，或者缓存中没有这条数据，那么大量的请求就会冲击到数据库上，而数据库的承压能力较为差劲，实际开发中一些热点的请求都不会通过数据库直接返回数据到客户端上，数据库一时间承受不住压力，就会导致数据库崩溃，就会严重的影响了用户的体验。

解决方案：

回填空值，在缓存库中设置key为null值，无论是系统崩溃还是不存在数据都会查到这个key，防止大量的请求进入到数据库，注意点：当数据库中key值新增了，那么就需要回填到缓存中，数据库更新后可以做一个异步线程处理回填空值。

使用布隆过滤器：原理就是快速的利用搞笑的算法检测key是否存在数据库中，不存在return就好了，存在的话就再去查数据库，刷新redis的key-value然后return，相当于再服务器和缓存中间加了一层过滤。

加锁处理：设置热点数据永不过期，然后在后端中加上互斥锁

Lock lock = new ReentrantLock();
    @Override
    public SysUser getHotInfo(int uid) throws InterruptedException {
        //从redis中查询数据
        SysUser user = redisCache.getCacheObject(String.valueOf(uid));
        //数据校验
        if (user==null){
            try {
                if (lock.tryLock()){
                    //去数据库查询数据
                    user = userMapper.selectByPrimaryKey((long)uid);
                    //校验
                    if (user!=null){
                        //存到缓存中
                        redisCache.setCacheObject(String.valueOf(uid),user);
                    }
                }else {
                    //睡一会再拿
                    Thread.sleep(100L);
                    user = userMapper.selectByPrimaryKey((long)uid);
                }
            }finally {
                //释放锁
                lock.unlock();
            }
        }
        return user;
    }

通常还会去做一些拦截处理：比如在接口请求过来的时候增加校验，一些参数，用户授权，不合法的话直接return，比如id基础校验，设置id<=0直接拦截

缓存雪崩

缓存雪崩就是同一时间缓存大面积失效，跟redis瞬间没有了一样，这么大的数量级冲击数据库，数据库会承受不了，它跟缓存击穿有点类似，但是不同的时缓存击穿大多数是单点击穿，而雪崩则是大面积瘫痪。

解决方案：

处理的方法是，批量添加数据到redis里时，把每个key的有效时间都加一个随机值，这样就可以保证数据不会同时间大面积失效了。

1	setRedis（key, value, time+Math.random()*10000）;

如果redis是集群部署，将热点的数据均匀分布在不同的redis库中能避免全部失效，或者设置热点数据永远不过期，有更新操作就更新缓存好了，比如一些电商首页的数据。

Redis和Memcached的区别

1、存储方式上：memcache会把数据全部存在内存之中，断电后会挂掉，数据不能超过内存大小。redis有部分数据存在硬盘上，这样能保证数据的持久性。

2、数据支持类型上：memcache对数据类型的支持简单，只支持简单的key-value，，而redis支持五种数据类型。

3、使用底层模型不同：它们之间底层实现方式以及与客户端之间通信的应用协议不一样。redis直接自己构建了VM机制，因为一般的系统调用系统函数的话，会浪费一定的时间去移动和请求。

4、value的大小：redis可以达到1GB，而memcache只有1MB。

Redis的淘汰策略

redis有六种淘汰策略

volatile-lru ：对已经过期的key-value集中对最近少用的（less recently used）的数据进行淘汰
volatile-ttl : 对已经过期的key-value集中对剩余时间较短的（time to live）的数据进行淘汰
volatile-random：对已经过期的key-value集中随机选择数据淘汰
allkeys-lru：从所有key-vlaue集中优先对最近少用的（less recently used）的数据进行淘汰
allkeys-random：从所有key-vlaue集中选择随机数据进行淘汰
noeviction:不淘汰策略，若内存超过最大，返回错误信息

补充：redis4.0加入了LFU（least frequency use）淘汰策略，volatile-lfu和allkeys-lfu，通过访问的频率，将访问频率最少的，最不经常使用的淘汰。

持久化

redis为了保持效率，数据缓存在内存中，但会周期性的把更新的数据写入磁盘或者把修改操作写入磁盘中，保证数据的持久化。redis持久化策略有两种：

RDB:以快照的形式直接将内存中的数据保存到drump文件中，定时保存，还有保存的策略。
AOF：把redis上所有的指令都存到一个aof文件中，命令几种存放。

Redis默认的持久化方式是RDB

RDB执行流程：当Redis需要做持久化时，Redis会fork一个子进程，子进程将数据写到磁盘上一个临时RDB文件中。当子进程完成写临时文件后，将原来的RDB替换掉，这样的好处是可以copy-on-write。

RDB的优点：这种文件非常适合用于备份：比如，你可以在最近的24小时内，每小时备份一次，并且在每个月的每一天也备份一个RDB文件。这样的话，即使遇上问题，也可以随时将数据集还原到不同的版本。RDB非常适合灾难恢复。

RDB的缺点：如果你需要尽量避免在服务器故障时丢失数据，那么RDB不合适你。

AOF的优点：让redis变得非常耐久。可以设置不同的fsync策略，aof的默认策略是每秒钟fsync一次，在这种配置下，就算发生故障停机，也最多丢失一秒钟的数据。缺点是对于相同的数据集来说，AOF的文件体积通常要大于RDB文件的体积。根据所使用的fsync策略，AOF的速度可能会慢于RDB。

数据库备份和灾难恢复

定时生成RDB快照非常便于进行数据库备份，并且RDB恢复数据集的速度也要比AOF恢复的速度快。当然了，redis支持同时开启RDB和AOF，系统重启后，redis会优先使用AOF来恢复数据，这样丢失的数据会最少。

主从复制

Redis主从复制作用

1.一个master可以有多个slave

2.除了多个slave连到相同的master外，slave也可以连接其他slave形成图状结构

3.主从复制不会阻塞master。也就是说当一个或多个slave与master进行初次同步数据时，master可以继续处理client发来的请求。相反slave在初次同步数据时则会阻塞不能处理client的请求。

4.主从复制可以用来提高系统的可伸缩性,我们可以用多个slave 专门用于client的读请求，比如sort操作可以使用slave来处理。也可以用来做简单的数据冗余

5.可以在master禁用数据持久化，只需要注释掉master 配置文件中的所有save配置，然后只在slave上配置数据持久化。

6.可以用于读写分离和容灾恢复。

redis主从复制常用的几种方式

一主二仆 A(B、C) 一个Master两个Slave
薪火相传（去中心化）A - B - C ，B既是主节点（C的主节点），又是从节点（A的从节点）
反客为主（主节点down掉后，手动操作升级从节点为主节点） & 哨兵模式（主节点down掉后，自动升级从节点为主节点）

Redis主从复制原理

Ø 当设置好slave服务器后,slave会建立和master的连接，然后发送sync命令。

Ø Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令，在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步。

Ø 全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。（第一次全量）

Ø 增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步。（之后增量）

Ø 但是只要是重新连接master,一次完全同步（全量复制)将被自动执行。

哨兵模式

Redis主从复制会出现什么问题？

一旦主节点宕机，从节点晋升为主节点，同时需要修改应用方的主节点地址，还需要命令所有从节点去复制新的主节点，整个过程需要人工干预。
主节点的写能力受到单机的限制。
主节点的存储能力受到单机的限制。
原生复制的弊端在早期的版本中也会比较突出，比如：redis复制中断后，从节点会发起psync。此时如果同步不成功，则会进行全量同步，主库执行全量备份的同时，可能会造成毫秒或秒级的卡顿。

解决方案：使用哨兵模式配置主从。

Redis的Sentinel系统可以用来管理多个Redis服务器，该系统可以执行以下四个任务：

监控：不断检查主服务器和从服务器是否正常运行。
通知：当被监控的某个redis服务器出现问题，Sentinel通过API脚本向管理员或者其他应用程序发出通知。
自动故障转移：当主节点不能正常工作时，Sentinel会开始一次自动的故障转移操作，它会将与失效主节点是主从关系的其中一个从节点升级为新的主节点，并且将其他的从节点指向新的主节点，这样人工干预就可以免了。
配置提供者：在Redis Sentinel模式下，客户端应用在初始化时连接的是Sentinel节点集合，从中获取主节点的信息。

哨兵的工作原理是什么？

1、每个Sentinel节点都需要定期执行以下任务：每个Sentinel以每秒一次的频率，向它所知的主服务器、从服务器以及其他的Sentinel实例发送一个PING命令。（如上图）

2、如果一个实例距离最后一次有效回复PING命令的时间超过down-after-milliseconds所指定的值，那么这个实例会被Sentinel标记为主观下线。（如上图）

3、如果一个主服务器被标记为主观下线，那么正在监视这个服务器的所有Sentinel节点，要以每秒一次的频率确认主服务器的确进入了主观下线状态。

4、如果一个主服务器被标记为主观下线，并且有足够数量的Sentinel（至少要达到配置文件指定的数量）在指定的时间范围内同意这一判断，那么这个主服务器被标记为客观下线。

5、一般情况下，每个Sentinel会以每10秒一次的频率向它已知的所有主服务器和从服务器发送INFO命令，当一个主服务器被标记为客观下线时，Sentinel向下线主服务器的所有从服务器发送INFO命令的频率，会从10秒一次改为每秒一次。

6、Sentinel和其他Sentinel协商客观下线的主节点的状态，如果处于SDOWN状态，则投票自动选出新的主节点，将剩余从节点指向新的主节点进行数据复制。

7、当没有足够数量的Sentinel同意主服务器下线时，主服务器的客观下线状态就会被移除。当主服务器重新向Sentinel的PING命令返回有效回复时，主服务器的主观下线状态就会被移除。

Redis哨兵的高可用

Redis Sentinel是一个分布式架构，包含若干个Sentinel节点和Redis数据节点，每个Sentinel节点会对数据节点和其余Sentinel节点进行监控，当发现节点不可达时，会对节点做下线标识。

如果被标识的是主节点，他还会选择和其他Sentinel节点进行“协商”，当大多数的Sentinel节点都认为主节点不可达时，他们会选举出一个Sentinel节点来完成自动故障转移工作，同时将这个变化通知给Redis应用方。

整个过程完全自动，不需要人工介入，所以可以很好解决Redis的高可用问题。