引言

自Redis入门篇过后，已经好久没更Redis了，接下来应该从实战篇，原理篇，面试篇几个层次来展开，本篇主要是实战篇环节，以问题展开，应对面试场景作答【melo称其为"手撕面答"】，尽量简短，某些部分可能不会进行详细介绍。

emmm，但后边有些部分还是干脆整合在一起了，可观性好一点，不至于看得一头雾水

本篇脑图速览

Redis限流是怎么做的？

固定窗口计数

固定窗口计数是指，假设我们的限流规则是：1min内最多只能访问10次，那么固定窗口就是固定了【 1min-2min】这个窗口内，只能有10次访问 ，相应的我们就要给这个窗口维护一个计数器。 为了节省空间，其实我们不需要维护一个个窗口，只需要维护当前访问时间所在的窗口即可，以及对应的计数器，当新的访问到达了下一个窗口时，则计数器重置即可。

redis实现

用redis的话，由于有过期机制，其实设置1min过期，就可以实现计数器重置的效果了

• redis设置一个名为qps的key，val用来计数，1min过期即可

//原子自增类
RedisAtomicInteger redisAtomicInteger = new RedisAtomicInteger(redisKey, redisTemplate.getConnectionFactory());
//先自增
int qps = redisAtomicInteger.getAndIncrement();
//若是第一次访问
if(qps==0){
    //设置1min过期
	redisAtomicInteger.expire(1, TimeUnit.MINUTES);
}
if(qps>10){
	throw new RuntimeException("qps超过阈值");
}
复制代码

存在的问题

由于是固定窗口，那其实存在窗口临界问题，比如用户可以在【1.5-2】这段区间访问10次，【2-2.5】这段区间也访问10次，这样就变成了1min内其实可以访问20次！看起来破坏了我们的限流规则，但由于我们是固定窗口计数，到达2的时候已经重置计数器了。

滑动窗口计数

假设我们的限流规则是：1min内最多只能访问10次，那么滑动窗口呢就是会根据你访问进来的时间，以访问时间作为区间末尾，当前时间-1min作为区间头部，相当于窗口一直在往右滑动，这样其实就能在一定程度上解决我们刚才提到的窗口临界问题

• 当访问时间为2.5的时候，此时对于的窗口是【1.5-2.5】，计数器都能正确计数

实现

要获得一段区间，并且按时间排序，我们可以想到用ZSet来实现，能按区间查询出【当前访问时间-1min，当前访问时间】这段区间的计数

//interMills为限流时间，也就是我们这里的1min
Long count = redisTemplate.opsForZSet().count(redisKey, currentTimeMillis - interMills, currentTimeMillis);
复制代码

存在的问题

其实我们只是以更小的窗口大小去移动这个区间罢了，固定窗口计数是以1min为单位去移动，滑动窗口是以1s为单位去移动，后者出现窗口临界问题的概率更小，但依然是可能出现的，比如：

1. 一开始是【1-2min】这段区间，下一秒会移动为【1min1s - 2min1s】，如果此时有人在1min这一刻，访问了10次，然后下一秒又进入下一个区间了，计数重置，在1min1s这一刻又访问了10次，依旧会出现窗口临界问题，1min内访问次数达到了20次

经评论区的小伙伴提问过后，发现其实滑动窗口算法是能够解决临界窗口问题的，当初学习时，可能只看了片面的资料，或者那个资料的实现方式不是用ZSet，而是用其他，以起点为首的区间去计算也有可能。

不过至少可以确定的是，本文用ZSet实现，以当前访问时间为区间末尾的话，确实是不会发生临界窗口问题的，非常感谢两位掘友：[吃西瓜啊] && [kb啵]

窗口临界问题小结

其实窗口临界问题，就是在即将被移出窗口的这段区间内，可能一次性访问量达到了我们的阈值，而由于要移出窗口了，计数又将重置了，所以这些访问量就相当于不会被后续统计到，那么后续再次超过阈值，就变成双倍阈值了。

漏桶算法

不限制流入，只限制流出的速率 -- 以一定的速率，去获取桶中的请求

水(请求)先进入到漏桶里，漏桶以一定的速度出水【从下方漏出去】(接口有响应速率)，当水流入速度过大会直接溢出【从上边移除】（访问频率超过接口响应速率)，然后就拒绝请求，可以看出漏桶算法能强行限制数据的传输速率。

但无法应对突发流量，因为流出的速率，是限死的

漏桶由于是底部有一个破洞，以固定速率流出请求，顶部用来接收请求，所以其实可以用一个双端队列来实现

1. 收到请求时，放入队列中【后台线程以固定的速率去removeLast，处理请求】
2. 若队列满了，则请求就被丢弃。

令牌桶算法

限制流入，不限制流出 -- 以一定速率，去生成桶中的请求【令牌】

先有一个木桶，系统按照固定速度，往桶里加入Token，如果桶已经满了就不再添加。 当有请求到来时，会各自拿走一个Token，取到Token 才能继续进行请求处理，没有Token 就拒绝服务。

支持突发流量

如果一段时间没有请求时，桶内就会积累一些Token，下次一旦有突发流量，只要Token足够，也能一次处理，所以令牌桶算法的特点是_允许突发流量_

Redis实现

定时任务

// 每个请求都需要获取令牌
public Response limitFlow(Long id){
        Object result = redisTemplate.opsForList().leftPop("flowList");
        if(result == null){
            return Response.ok("当前令牌桶中无令牌");
        }
        return Response.ok(articleDescription2);
}
复制代码

再依靠Java的定时任务，定时往List中rightPush令牌

// 10S生成一个令牌，放入令牌桶中，使用UUID保证唯一性
    @Scheduled(fixedDelay = 10_000,initialDelay = 0)
    public void setIntervalTimeTask(){
        redisTemplate.opsForList().rightPush("flowList",UUID.randomUUID().toString());
    }
复制代码

懒惰更新令牌

1. 记录上一次访问时间，当新的访问进来时，令牌数量+= (新访问时间-上一次访问时间)×每秒生成的令牌数量
2. 然后判断请求数是否大于令牌数量 大于：则拒绝掉 小于：则减掉相应的令牌数量即可

令牌桶与漏桶相比

• 令牌桶限制的是平均流入速率（允许突发请求，只要有令牌就可以处理，支持一次拿3个令牌，4个令牌），并允许一定程度突发流量；
• 漏桶限制的是常量流出速率（即流出速率是一个固定常量值，比如都是1的速率流出，而不能一次是1，下次又是2），从而平滑突发流入速率；
• 令牌桶允许一定程度的突发，而漏桶主要目的是平滑流入速率；

除了用Redis限流，还能否在其他层面做限流？

Nginx 两种限流方式

控制速率

ngx_http_limit_req_module 模块提供了漏桶算法(leaky bucket)，可以限制单个IP的请求处理频率。

正常限流：

http {
    limit_req_zone 192.168.1.1 zone=myLimit:10m rate=5r/s;
    }
    
server {
    location / {
        limit_req zone=myLimit;
        rewrite / http://www.hac.cn permanent;
    }
}
复制代码

参数解释： key: 定义需要限流的对象。 zone: 定义共享内存区来存储访问信息。 rate: 用于设置最大访问速率。 表示基于客户端192.168.1.1进行限流，定义了一个大小为10M，名称为myLimit的内存区，用于存储IP地址访问信息。rate设置IP访问频率，

rate=5r/s表示每秒只能处理每个IP地址的5个请求。Nginx限流是按照毫秒级为单位的，也就是说1秒处理5个请求会变成每200ms只处理一个请求。如果200ms内已经处理完1个请求，但是还是有有新的请求到达，这时候Nginx就会拒绝处理该请求。

突发流量限制访问频率

上面rate设置了 5r/s，如果有时候流量突然变大，超出的请求就被拒绝返回503了，突发的流量影响业务就不好了。

这时候可以加上burst 参数，一般再结合 nodelay 一起使用。

server {
  location / {
    limit_req zone=myLimit burst=20 nodelay;
    rewrite / http://www.hac.cn permanent;
  }
}
复制代码

没有nodelay的话：

若一瞬间超出设置的每秒处理量时，允许超出 20 个请求，这20个请求会阻塞排队等待处理【相当于继续加入桶的头部，等待定时任务从队尾remove】，多于20的部分则被丢弃

有nodelay的话：

burst=20 nodelay 表示这20个请求立马处理，不能延迟，相当于特事特办。

我们的业务有时可以一瞬间就处理完很多个请求，这种情况下配置nodelay就可以一瞬间处理完这些突发流量，而不用继续阻塞排队等待处理。

burst+delay

在峰值快速处理的例子中，当接收到超出限定速率的请求时，可以一定程度上快速处理，但系统的承受能力毕竟是有限的，所以burst的大小会受限于系统的承受能力。假如系统能承受的最大瞬间并发量为2000，但外部短时间内请求峰值为3000，那么这多出的这1000个请求就必须丢弃吗？有没有可能快速处理2000个请求，剩下1000个堵塞等待后续处理呢？

• 可以通过Nginx的 delay=? 配置来实现。这个配置表示在超出限定速率的请求中，超过多少个请求之后需要被延时处理，没有超过delay值的请求，无需等待。nodelay其实是delay的一种特殊情况，表示所有请求都无需等待，相当于delay的值等于burst。

配置示例：

http {
    limit_req_zone $binary_remote_addr zone=one:10m rate=12r/m;
    server {
        location / {
            limit_req zone=one burst=10 delay=8;
        }
}
复制代码

delay=8表示超出限定速率的8个请求可以被快速处理，再超过的（当然也必须小于burst）就必须阻塞等待。

这意味着，如果从一个给定 IP 地址发送 21 个请求，Nginx 会立即将第一个请求发送到上游服务器群，然后将余下 20 个请求放在队列中。然后每 100 毫秒转发一个排队的请求，只有当传入请求使队列中排队的请求数超过 20 时，Nginx 才会向客户端返回 503。

相当于20之后的是特权？前20又重新排队吗，什么意思

控制并发连接数

ngx_http_limit_conn_module 提供了限制连接数功能。

limit_conn_zone $binary_remote_addr zone=perip:10m;
limit_conn_zone $server_name zone=perserver:10m;

server {
  ...
    limit_conn perip 10;
  limit_conn perserver 100;
}
复制代码

limit_conn perip 10 作用的key 是 $binary_remote_addr，表示限制单个IP同时最多能持有10个连接。

limit_conn perserver 100 作用的key是 $server_name，表示虚拟主机(server) 同时能处理并发连接的总数。

注：limit_conn perserver 100 作用的key是 $server_name，表示虚拟主机(server) 同时能处理并发连接的总数。

RocketMQ削峰限流

这个跟我们本文关联性不大，我们后边再单独出一篇详细讲讲

Redis如何实现延时队列？

String设置key，过期监听触发事件

比如3点下单，5点需要执行定时任务发通知，此时设置一个key，过期时间为距离定时任务执行的时间【5-2】= 2h 等到该key即将过期时，redis中有一个监听机制，key过期时可以触发自定义事件，我们可以在代码里对不同key执行不同的操作，所以key过期的时候，触发自定义事件：根据key的内容去发定时任务通知就好了

存在问题

redis中的key过期策略，决定了定时任务不一定能准点执行

1. 主动访问键的时候，发现过期时删除
2. 后台线程定时扫描，发现过期时删除

所以如果我们长时间没有访问这个key，后台线程也没扫描到的话，这个key本质上是还未过期的，不会触发过期事件的

Redis 从未保证会在设定的过期时间立即删除并发送过期通知。实际上，过期通知晚于设定的过期时间数分钟的情况也比较常见。 此外键空间通知采用的是发送即忘（fire and forget）策略，并不像消息队列一样保证送达。当订阅事件的客户端会丢失所有在断线期间所有分发给它的事件。

所以，这个方案其实很少人使用。。

ZSet存储定时任务

延时队列其实有很多种实现的方式，比如RocketMQ本身支持发送延迟消息，但RocketMQ支持的延迟等级有限，自定义程度不高，比如我抢到了一个场地之后，要设置场地时间结束后，修改订单状态为【已结束】，这时RocketMQ就没法精准的设置一个定时任务的时间。

于是可以用Redis中的ZSet数据结构，以任务的时间作为作为score进行排序，按时间先后进行排序，后台再开启定时任务，定时利用 zrangebysocre 查询符合条件的所有待处理的任务即可

本地做定时任务不可以吗？为什么非得引个Redis

也可以实现，但一旦项目重新部署，那么JVM内存里边存放的定时任务也都会随之丢弃，相当于没有一个持久化的媒介。

当然，延时队列还有很多种实现方式，基于rabbitmq，时间轮算法等，这里先不深入

Redis如何实现布隆过滤器？

用BitMap可以实现，这里具体还是说说布隆过滤器的原理和优缺点即可，具体实现是类似的

布隆过滤器的实现原理？

(布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的**二进制向量(位图)**和一系列随机映射函数（哈希函数）。

布隆过滤器可以用于检索一个元素是否在一个集合中：

1. 如果布隆过滤器判断存在，则放行【可能会误判】，后续过程跟普通查询redis过程是一样的
2. 不存在，则直接返回，不走redis

将所有可能存在的数据哈希到一个足够大的bitmaps中，一定不存在的数据会被这个bitmaps拦截掉，从而避免了对底层存储系统的查询压力。

• 它的优点是空间效率和查询时间都远远超过一般的算法
• 缺点是有一定的误识别率和删除困难。

误判原因在于：布隆过滤器走的是哈希思想，只要哈希思想，就可能存在哈希冲突，数据x和数据y的哈希结果一样，如果只有x，判断y是否存在的时候，由于跟x的哈希值一样，导致布隆过滤器误以为y也存在

布隆过滤器的数据结构

布隆过滤器由「初始值都为 0 的位图数组」和「 N 个哈希函数」两部分组成。当我们在写入数据库数据时，在布隆过滤器里做个标记，这样下次查询数据是否在数据库时，只需要查询布隆过滤器，如果查询到数据没有被标记，说明不在数据库中。

布隆过滤器会通过 3 个操作完成标记：

• 第一步，使用 N 个哈希函数分别对数据做哈希计算，得到 N 个哈希值；
• 第二步，将第一步得到的 N 个哈希值对位图数组的长度取模，得到每个哈希值在位图数组的对应位置。
• 第三步，将每个哈希值在位图数组的对应位置的值设置为 1；

举个例子，假设有一个位图数组长度为 8，哈希函数 3 个的布隆过滤器。

在数据库写入数据 x 后，把数据 x 标记在布隆过滤器时，数据 x 会被 3 个哈希函数分别计算出 3 个哈希值，然后在对这 3 个哈希值对 8 取模，假设取模的结果为 1、4、6，然后把位图数组的第 1、4、6 位置的值设置为 1。当应用要查询数据 x 是否数据库时，通过布隆过滤器只要查到位图数组的第 1、4、6 位置的值是否全为 1，只要有一个为 0，就认为数据 x 不在数据库中。

注意：此处为什么需要3个hash函数？

若只有1个hash函数，冲突的概率是很大的，都hash到同一个位置，导致误判的概率很大 因此使用多个hash函数，hash到多个位置，只有这几个位置都是1，才说明x存在，误判的概率会降低些

布隆过滤器由于是基于哈希函数实现查找的，高效查找的同时存在哈希冲突的可能性，比如数据 x 和数据 y 可能都落在第 1、4、6 位置，而事实上，可能数据库中并不存在数据 y，存在误判的情况。

所以，查询布隆过滤器说数据存在，并不一定证明数据库中存在这个数据，但是查询到数据不存在，数据库中一定就不存在这个数据。

本质上是一个很大的位图，存储值的时候，用多个hash函数计算出他要存储的位置，比如x，对应hash后的结果是1,2,4， 把这几个位置标记为1。

查询的时候，对x做多次hash，只有所有hash后的位置标记位都是1，才可能存在

• 若有一个为0，则肯定不存在。

为什么是可能存在？

因为可能x已经不在缓存里了，此时又进来了一个y，y的hash结果跟x一模一样，此时会出现误判。

布隆过滤器为什么不好删除元素？

比如现在要删除x，对x做hash，找到了他的存储位置分别是【1,3,9】，我们如果直接把这三个位置改为0，可能会导致“删除”了其他元素

• 比如y他的存储位置是【2,3,8】，x跟y共用了3这个位置，把3这个位置改为0，会导致y也被“删除”了，根据上边的原理发现不是全1了

如何解决呢？

最简单的做法就是加一个计数器，就是说位数组的每个位如果不存在就是0，存在几个元素就存具体的数字，而不仅仅只是存1。

那么这就有一个问题，本来存1，一位就可以满足了，但是如果要存具体的数字，可能需要更多的位数，所以带有计数器的布隆过滤器会占用更大的空间。

幂等处理怎么用Redis实现？

比如我们有一个支付接口，一个订单号只能被支付一次

1. 客户端先调用获取token的接口，后台生成一个token，返回给前端，同时存储在redis里边，设置一定时间过期
2. 客户端带上token，调用支付接口，后台判断是否有这个token，有则说明是第一次访问 删除token 然后执行业务
3. 第二次访问时，检测到没有token，则不允许执行，确保幂等性

?♂?先删除token，还是先执行业务？

1. 先删除token，若后续执行业务期间失败了，则直接第二次点击按钮，调用支付接口时，由于没有刷新页面，前端还存储了刚才那份token，由于后台没有这个token，就会一直失败

• 所以此时需要：客户端主动刷新页面，删除掉前端这个token，重新完成提交这个过程，重新调用获取token接口，再走一遍流程

2. 先执行业务，再删除token，但此时若没有加锁的话，其他线程调用接口时，由于A线程还没执行完业务，redis里边的token还未删掉，那么B线程调用支付接口时，也会查到还有token，也能够去执行业务，这样就破坏我们的幂等性了。

• 所以如果采用这种策略的话，需要加锁，保证A线程执行完业务，删掉token之后，其他线程才能调用这个支付接口。

如此来看，还是第一种方式更优，业务执行出错的话，前端重新刷新页面也能再次成功。

此处有待补充，一些细节问题可能还没有充分考虑到，实现幂等，也还有很多种方式，后边再来深入分析

Redis的incr可以做什么？

签到等需要计数的序列号

每天签到的序号，设置1天过期，第一次签到时便是从0开始自增

记录登录失败次数防爆破

记录某个用户登录失败的次数，防爆破攻击，1min内失败次数超过5次则限制10min内不允许再次调用登录接口

数据结构设计

1. 一个login_error的键来记录失败次数，key是 login_error:+账号+ip，val是失败次数， 1分钟过期

每次登录失败，就会val++

2. 当失败次数超过5次，则设置一个 bank 的键，key是 login_bank:+账号+ip，val是1, 10分钟过期 并把上边的login_error键删除

注意redis里边key的前缀用 ：来分隔，可以在图形化界面实现文件夹管理的形式

登录失败的逻辑

1. 先判断是否有 login_bank:+账号+ip的key，有则直接限制登录 没有，若登录失败，则 login_error:+账号+ip的 val++， 若val达到5了，则设置一个 login_bank:+账号+ip的key，10min过期 并删除掉 login_error:+账号+ip

当然，也可以不用设置两个key，单纯一个key就可以了，发现val>5了，则延长这个key的时间为10min也是可以的

作者：Melo_
链接：https://juejin.cn/post/7158001985046708260