Redis总结

1. Redis 基本介绍、安装。

   Redis用来解决用户量增大，请求增大，数据库压力过大，多台服务器之间数据不同，传统锁失效

   Redis是非关系型数据库，NoSQL，以key-value形式存储

   linux启动，src/redis-server redis.conf 进入 src/redis-cli -a 123 关闭 shutdown 退出 exit

五种基本数据类型(5中value)：string/Hash/List/set/zset。

1. 五种基本数据类型(5中value)：string/Hash/List/set/zset。

   因为是key-value形式存储，所以相同的key,value会被覆盖

   key-string:一个key对应一个值，最常用的，用于存储一个值

   key-hash: 一个key 对应一个Map，一般是存对象

   key-list： 一个key 对应一个列表，用list实现栈和队列结构

   key-set：一个key 对应一个集合，求交集，差集和并集

   key-zset:一个key 对应一个有序的集合，排行榜，积分存储等跟有序有关的操作

set类型

存数据，set k1 v1 批量存,mset k1 v1 k2 v2 …

查数据，get k1 批量查，mget k1 k2 …

自增命令 incr key 或incrby key 自增的值 自减 decr key

设置值同时设置生存时间 setex key second value

查看对应value的长度 strlen key

hash类型 类似set

存数据，hset 对象名 属性 属性值 批量存,hmset k1 field value k2 f2 v2 …

查数据，hget k1 f1 批量查，hmget k1 f1 k2 f2 …

List类型

存数据（从左侧插入，从右侧插入数据）

lpush k1 v1 v2 v3… rpush k1 v1 v2 …

set类型

存数据 sadd k1 v1 v2 …

获取数据 smembers k1

zset的常用的命令

添加数据(score必须是数值。member不允许重复的)

zadd key score member score member

查看member 的分数

zscore key member zrange key start stop byscore [withscores]

key的常用命令

查看所有的key keys pattern

查看key是否存在（1 存在，0不存在），exists key

设置key的生存时间 expire key second 或者pexpire key milliseconds

查看 key 生存的时间 （-2 表当前key 不存在，-1 表此key 没有设置生存时间，正数，具体时间）

移除key 的生存时间（1 移除成功，0 表key不存在生存时间或key不存在）

库 的操作命令

flushall 清空全部数据库

Redis事务一般不用，和正常事务不同，需要监听

Redis查数据先查生存时间，如果已经过期就会删除当前key,然后返回空值

Java 操作 redis

1. Java 操作 redis

   1. jedis 导入dedis依赖,创建Redis连接，Jedis jedis = new Jedis(“192.168.199.109”,6379);

      后面可直接创建连接池，管道，调用set方法，get方法设置获取值

      和数据库类似，从连接池里拿一个连接，在对数据库操作，Reids只是连接池，连接名字不一样，对Redis操作

      导入spring-context依赖 byte[]形式，fastjson依赖 string形式

   2. spring data redis（springboot2.0之后底层就是 lettuce，之前是 jedis）

   3. spring cache。五个注解 eanblecache cacheput cachedelete cacheconfig

      在配置文件加缓存策略名 spring.cache.cache-names=c1，在启动类加@EnableCaching 在业务层service包下加对应的注解会自动在redis缓存上操作数据，删除数据是先数据库在删缓存

两种数据持久化方案：（备份）

1. 两种数据持久化方案：（备份）

   1. RDB（快照持久化）。快照备份，二进制文件，内容无法理解，可以设置多少秒内多少次操作备份一次

   2. AOF（AOF持久化）。文本文件，能理解内容，一秒备份一次，最多丢失1秒的数据

      建议同时开启AOF和RDB ，如果同时开启了AOF和RDB持久化，那么在Redis宕机重启之后，需要加载一个持久化文件，优先选择AOF文件。如果先开启了RDB，再次开启AOF，如果RDB执行了持久化，那么RDB文件中的内容会被AOF覆盖掉。

      两种持久化方案都可以看作是非关系型数据库

主从搭建

1. 主从搭建

   1. 主从。一个主机带一个或两个从机，一个从级再带一个从机

      作用：主从复制可以在一定程度上扩展redis性能，redis的主从复制和关系型数据库的主从复制类似，从机能够精确的复制主机上的内容。实现了主从复制之后，一方面能够实现数据的读写分离，降低master的压力，另一方面也能实现数据的备份。

      原理：主机有一个replication ID 记录更新的数据,从机发送sync命令，主机传所有数据给从机，后面只传更新数据

          1.slave启动成功连接到master后会发送一个sync命令。  
          2.Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令。  
          3.在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步。  
          4.全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。  
          5.增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步。  
          6.但是只要是重新连接master,一次完全同步（全量复制)将被自动执行。  
      1
      2
      3
      4
      5
      6

      主机可以读写，从机只可以读，一般主机写，从机读。

      作用：主从模型可以提高读的能力，在一定程度上缓解了写的能力。因为能写仍然只有Master节点一个，可以将读的操作全部移交到从节点上，变相提高了写能力。即Redis主从提高了读写能力

哨兵模式

​ 哨兵模式。主机断开连接，sential自动投票在从机中选出一个主机

哨兵模式 sential.conf 最后的参数，表得票数多少才能称为主机，一个sential 只能投一票，会根据从机的性能投，提高了稳定性

redis 集群

redis 集群。16384 个slots(hash槽)，节点平均分配区间，存key-value，根据key得到一个hash值，hash值与16384取于，保证了在1到16284之间，根据结果存在相应的节点，

加一个节点，需要从原先节点分配，删除需要将节点转回原先节点，一个主机节点挂点，整个集群无法使用，从机节点挂掉影响不大。

Redis 集群运行原理如下：

1. 所有的 Redis 节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽

2. 节点的 fail 是通过集群中超过半数的节点检测失效时才生效

3. 客户端与 Redis 节点直连,不需要中间 proxy 层，客户端不需要连接集群所有节点，连接集群中任何一个可用节点即可

4. Redis-cluster 把所有的物理节点映射到 [0-16383]slot 上,cluster (簇)负责维护 node<->slot<->value。Redis 集群中内置了 16384 个哈希槽，当需要在 Redis 集群中放置一个key-value 时，Redis 先对 key 使用 crc16 算法算出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，Redis 会根据节点数量大致均等的将哈希槽映射到不同的节点。

   作用：所谓的集群，就是通过添加服务器的数量，提供相同的服务，从而让服务器达到一个稳定、高效的状态。单个redis存在不稳定性。当redis服务宕机了，就没有可用的服务了。 单个redis的读写能力是有限的。redis集群是强化了redis的读写能力，增加了Redis服务的稳定性

redis 缓存穿透：布隆过滤器。

每一个布隆过滤器有一个位数组和几个不同hash函数
add操作， 根据几个不同的hash函数给元素进行hash运算一整个索引值，拿到这个索引值之后，对位数组的长度进行取模运算，得到一个位置，每个hash函数都会得到一个位置，将位数组中对应的位置设置为1，这样就完成了添加操作

exists 操作，判断是否元素是否存在，先对元素进行hash运算，将运算的结果和位数组取模，然后去对应的位置查看是否为1，
如果每个位置都是1，表元素可能存在（因为这个1可能是其他元素存进来的），如果对应的位置有0，则这个元素一定不存在

位数组越大，误判概率越小，占用的存储空间越大

docker 安装redisbloom,配置好reids,conf，启动redis就会启动布隆过滤器

导入jrebloom依赖，new一个Client，Client调用add方法存储数据和访问数据前用exists方法检查数据

将数据存在布隆过滤器中（不存在redis中了），请求来了，首先去判断数据是否存在，如果存在，再去数据库中查询，否则就不去数据库中查询。

redis分布式锁

还是和单体服务差不多，都是加锁，单体服务是线程加锁，其他线程就需要等待锁释放，分布式锁，就是多服务，线程锁对多服务多个请求无效，所以锁的是单个服务单个请求，当A服务在操作b数据，其他服务请求想操作b数据必须等A服务操作完才能操作

第一种用jedis自带的方法实现另类锁，一个服务能执行setnx存k1，就可以执行其他操作，其他服务只能等待该服务释放setnx,删除k1

jedis.setnx(“k1”,“v1”)；第一次执行结果为1，之后执行有k1存在返回结果为0，

long setnx =jedis.setnx("k1","v1");
if(setnx==1){
    //说明没有人占位，可以操作
    //给锁添加一个过期时间，防止应用在运行过程中抛出异常导致锁无法及时得到释放
    //jedis.expire("k1",5L);
    xxxx操作
    //做完之后，释放k1
    jedis.del("k1");
}else{
    //有人操作，稍后再试
    System.out.println("没拿到锁");
}
1
2
3
4
5
6
7
8
9
10
11
12

解决5秒之后还没完成操作，锁就别释放，用lua脚本

public class LockDemo01 {
    public static void main(String[] args) {
        new Redis().execute(jedis -> {
            SetParams setParams =new SetParams()
                    //相当于执行setnx
                    .nx()
                    //相当于执行setex
                    .ex(5L);
            String value = UUID.randomUUID().toString();
            String set = jedis.set("k1", value, setParams);
            if(set != null && "OK".equals(set)){
                //没人占位
                jedis.set("name","javaboy");
                String name = jedis.get("name");
                System.out.println(name);
                jedis.evalsha("b8059ba43af6ffe8bed3db65bac35d452f8115d8", Arrays.asList("k1"),Arrays.asList(value));
            }else {
                //有人占位，停止/暂缓 操作
                System.out.println("没拿到锁");
            }

        });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

Redisson实现分布式锁（推荐）

导入依赖，redisson

获取redis连接，获取RedissonClient,获取锁，设置锁的超时时间，得到写业务

package com.qfedu;

import org.redisson.Redisson;
import org.redisson.api.RBucket;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;

public class LockDemo02 {
    public static void main(String[] args) {
        Config config =new Config();
        //配置 Redis 基本连接信息
        config.useSingleServer().setAddress("redis://192.168.183.128:6379").setPassword("123");
        //获取一个RedissonClient 对象
        RedissonClient redisson = Redisson.create(config);
        //获取一个锁对象实例
        RLock lock = redisson.getLock("lock");
        try {
            //获取锁
            //第一个参数是获取锁的等待时间 第二个参数是锁的超时时间
            boolean b = lock.tryLock(5L, 10L, TimeUnit.SECONDS);
            if(b){
                //获取到锁，开始写业务
                RBucket<Object> bucket = redisson.getBucket("k1");
                bucket.set("zhangsan");
                Object o = bucket.get();
                System.out.println("o = " + o);

            }else {
                System.out.println("没拿到锁");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //释放锁
            lock.unlock();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

redis 限流 漏斗算法

思路：一个令牌桶，请求拿到令牌才能通过，令牌桶有固定数量令牌，同时再不断以自定义的速度生成令牌，就可以达到限流效果

给一个令牌桶，linux安装redis-cell ,

wget https://github.com/brandur/redis-cell/releases/download/v0.2.4/redis-cell v0.2.4-x86_64-unknown-linux-gnu.tar.gz 
1

修改redis.conf文件，加载额外的模块，然后再启动redis

redis 启动成功后，如果存在 CL.THROTTLE 命令，说明 redis-cell 已经安装成功了。 CL.THROTTLE 命令一共有五个参数

1. 第一个参数是 key 2. 第二个参数是令牌桶容量 3. 令牌产生个数 4. 令牌产生时间 5. 本次取走的令牌数

执行完成后，返回值也有五个： 1. 第一个 0 表示允许，1表示拒绝 2. 第二个参数是令牌桶的容量 3. 第三个参数是当前桶内剩余的令牌数 4. 失败时表还需要等待多少秒可以有足够的令牌 5. 表预计多少秒后令牌桶会满

java代码实现

先启动redis, 进入redis,执行 CL.THROTTLE k2 100 1 1 10

定义一个Redis令牌桶指令接口，将@Command中的语句传到redis中执行

public interface MyRedisCommand extends Commands {
    @Command("CL.THROTTLE ?0 ?1 ?2 ?3 ?4")
    List<Object> throttle(String key,Long init,Long count,Long period,Long quota);
}

1
2
3
4
5

直接调用

public class RedisCellDemo {
    public static void main(String[] args) {
        RedisClient redisClient = RedisClient.create("redis://123@192.168.183.128");
        StatefulRedisConnection<String, String> connect = redisClient.connect();
        RedisCommandFactory factory = new RedisCommandFactory(connect);
        MyRedisCommand commands = factory.getCommands(MyRedisCommand.class);
        //执行指令，
        List<Object> list = commands.throttle("k2", 100L, 1L, 1L,10L);
        //输出结果
        System.out.println(list);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13

redis 幂等式处理接口

幂等性：以 POST 请求为例，多次请求同一个接口，且参数相同，对数据库而言，只会添加一条记录。

思路：第一种，把所有请求参数合到一起生成一个hash值，存入数据表中，同时建立一个唯一索引，第二次相同请求存不进去就表示请求无效

第二种，首先，再发送Post请求前，先从服务端获取一个token,把它存入Redis中，前端发送请求的时候，携带上token,服务端收到请求后，首先查看Redis 中是否存在token,如果存在，就让请求正常执行，同时移除token

1.服务端提供一个接口，这个接口返回一个token,uuid随机字符串，同时将token 存入Redis

2.客户端发送请求的时候，需要携带上token（（存在请求头中或者请求参数中））

3.服务端收到请求后，首先检查Redis 中是否存在该token

​ 1.存在：请求正常处理，同时移除掉Redis中的token

​ 2.不存在：说明该请求已经被处理过了，不在处理

此方法有一个弊端，一个请求需要请求两次，第一次来获取token,第二次在携带这个token去访问服务，（一般是有专门的页面，点击进入这个页面就获取token,然后在这个页面点击具体操作，这个操作请求就会携带token去访问服务）根据这个token来解决幂等性问题

代码流程，就是要自定义一个注解来区分需要解决幂等性问题的请求调用的服务，

用拦截器解析：在配置包中建一个WebMvcConfig 类，拦截过滤前端的请求，在Handler处理器处理前中，即创建一个类继承HandlerInterceptor类，获取所有controller包下的方法的注解，如果有一个注解等于自定义的注解，即表示该方法是需要处理幂等性的服务方法，调用token服务的验证方法进行token验证。

用aop解析：加aop依赖 ，设置切点，获取带有自定义注解的请求，然后就是token验证

关键是拦截请求，得到前端请求就随便搞了，aop或拦截器解析，定义一个全局异常类GloablException返回运行时异常的信息

使用场景：

session 共享

**在集群中，用 redis 做 session 共享。**是单点登录功能的原理

前端用户传来用户名，密码，session等信息，然后根据用户名密码生成key再从redis获取对应的数据判断是否相等，

相等再看用户的权限信息，选择对应的权限展示给用户，将key作为cookie写回游览器

登录成功，声名一个uuid作为key,用户对象作为value存入redis,返回key到controller

session共享：游览器通过Nginx请求转发到不同的服务器Tomcat,然后不同服务器Tomcat都把session放在Redis中，就实现了session共享。所以还需要安装nginx，用到反向代理

加入依赖SpringWeb,Spring Session,Spring Data Redis（Access+Driver）

配置文件

spring.redis.host=192.168.183.128
spring.redis.password=123
spring.redis.port=6379

server.port=8080

1
2
3
4
5
6

controller

@RestController
public class HelloController {

    @Value("${server.port}")
    Integer port;
    @GetMapping("/set")
    public String set(HttpSession session){
        session.setAttribute("name","zhangsan");
        return String.valueOf(port);

    }
    @GetMapping("/get")
    public String get(HttpSession session){
        return session.getAttribute("name")+":"+port;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

正反向代理

反向代理：给10086打电话，10086在转发给多个话务员的一个，你只要打10086，不是打给话务员，一个请求到服务器，服务器可以转到多个服务，就是反向代理，请求不在意是多个服务器的哪个，只要通过中间服务器就可得到服务响应就ok

正向代理：游览器无法直接访问谷歌，因为dns污染（访问谷歌域名时需要dns解析，dns污染胡乱解析就无法访问域名），可以直接访问谷歌ip地址就不用访问域名不需要解析，谷歌在全球就有很多的ip,ip用多了就会被封，但是不会封完，特殊时期就会封完，vpn就是一个服务器能通过ip访问谷歌，游览器通过访问vpn,跳转请求通过ip访问到谷歌，vpn服务器代理的就是游览器的请求，就是正向代理。多个请求到服务器，服务器通过固定的ip访问到一个谷歌，就是正向代理.谷歌服务器不在意哪个请求，只要把响应给中间服务器，中间服务器就能把响应给到具体的请求

1. Redis做延迟消息队列，用zset,zset中有score 是一个数字，我们可以设置消息的延迟时间为score

Redis问题

查询数据，先去redis中查，再去数据库中查

缓存穿透问题

原因： 查询的数据，Redis中没有，数据库中也没有
如果恶意请求，根据id 查数据，携带了大量数据库中也没有的id,多此执行这样的请求会把数据库拖垮，redis没有起到缓冲的作用
解决办法:

缓存空对象，对于每次没查找到的请求，设置一个空缓存

1.根据id查询时，如果id 是自增的，将id 的最大值放到Redis中，在查询数据库之前，直接比较一下id
2.如果id 不是整形，可以将全部的id放到set中，在用户查询之前，去set中查看一下是否有一个id（低效）
3.获取客户端的ip地址，可以将ip的访问添加限制

最佳解决是使用布隆过滤器

每一个布隆过滤器有一个位数组和几个不同hash函数
add操作， 根据几个不同的hash函数给元素进行hash运算一整个索引值，拿到这个索引值之后，对位数组的长度进行取模运算，得到一个位置，每个hash函数都会得到一个位置，将位数组中对应的位置设置为1，这样就完成了添加操作

exists 操作，判断是否元素是否存在，先对元素进行hash运算，将运算的结果和位数组取模，然后去对应的位置查看是否为1，
如果每个位置都是1，表元素可能存在（因为这个1可能是其他元素存进来的），如果对应的位置有0，则这个元素一定不存在

位数组越大，误判概率越小，占用的存储空间越大

缓存击穿问题

原因：缓存中的热点数据，突然到期了，造成了大量的请求都去访问数据库，造成数据库宕机

解决办法：

1. 在访问缓存中没有的时候，直接添加一个锁，只让几个请求去访问数据库，避免数据库宕机（会降低并发的性能）
   2.热点数据的生存时间去掉，变为永久存在

缓存雪崩问题

原因：当大量缓存同时到期时，最终大量的同时去访问数据，导致数据库宕机

解决：将缓存中的数据的生存时间，设置为30~60的一个随机时间

缓存击穿和雪崩，都是数据生存时间突然全部到期，导致大量请求去访问数据库从而导致数据库宕机，都可以设置一个随机生存时间来解决，不过击穿是热点数据到期，雪崩是普通数据到期，热点数据较少可以直接设置为永久存在

缓存倾斜问题

原因：热点数据放在一个Redis节点上，导致Redis节点无法承受住大量的请求，最终Redis宕机
解决：1.扩展主从架构，搭建大量的从节点，缓解Redis的压力
2.可以在Tomcat 中做JVM缓存，在查询Redis 之情，先去查询Tomcat 中的缓存

Redis 用Hash 存优惠券

Redis存热点数据，即经常使用又不轻易改变的数据，不重要的数据

利用redis4.x自身特性，LFU机制发现热点数据。实现很简单，只要把redis内存淘汰机制设置为allkeys-lfu或者volatile-lfu方式，再执行**./redis-cli --hotkeys**会返回访问频率高的key（即点击频繁的数据），并从高到底的排序,在设置key时，需要把商品id带上，这样就是知道是哪些商品了。然后将热点数据加一个属性，即点击值，根据点击值来降序排序，加载时就更快

用jedis 和数据库类似，从连接池中获取一个Redis 连接，连接上后，直接调用set,get方法，就可以查值了

用RedisTemplate,需要序列化，需要IO ,ObjectMapper om = new ObjectMapper();需要转Json,需要注解@Cacheable,@CachePut,@CacheEvict

        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
1
2
3
4
5
6

限流就是限制请求数QPS或者线程数，降级跳到就是fallback的接口实现函数，返回空值或者默认的值，或者一些友情提示，当前请求过多，请稍后再试，限流（hystrix,sentinal）和降级保证数据库不会死，可以处理2/5的请求，对用户来说就是多点击几次刷新页面就可以得到结果

Redis宕机 事发前 主从结构+哨兵 或者集群，事发时，本地缓存ehcach和限流降级，事发后就利用持久化，重启redis,恢复数据