相信各位对redis肯定是不陌生的，一个高吞吐量的内存型结构存储数据库。可用用于很多业务场景，能够有效的解决很多复杂的并发问题，分布式问题。

下面粘一下中文官网介绍：

关于解决对象共享问题，很多方式，通过一般的关系型数据库就可以(mysql)，但是相较而言，mysql关系型数据库和nosql数据库，两者读写效率也是不一样的，一个在硬盘上工作，一个在内存上工作，此就是差距；频繁的IO操作，大大降低了CPU性能；redis采用cache。完全不一样的性能。

用一组数据对比：

redis读写能力为2W/s

mysql读能力5K/s、写能力为3K/s

从数据角度来说，基本上是碾压性的。

可想，对于redis来说，很方便适用于各种高频读写内存的场景。然而，对于单机版的应用，我们直接将变量、对象放在主机的物理内存上就可以了，也不需要用到这种中间件。但是对于分布式的环境下。我们无法实现内存共享。必然是需要借助于redis这个中间件技术的。才可以实现多节点下的内存数据共享。

然后，对于我之前写的那篇文章可以知道，jvm中是存在原子性操作的方法的，cas

那么，即使在分布式环境下我们更需要考虑这个情况了。毕竟多节点，并发数就翻了几倍了。

so，我们强大的redis也考虑到这个了。

于是有了Redisson框架

可以这么讲，jdk中的juc包提供的是单机版的并发业务。那么Redisson基本是基于juc实现的分布式的业务

此处直通车Redisson官方文档

我们可以看到很多基于分布式的封装

很多分布式锁的实现，基本满足了所有业务场景

是的，我们今天讲的就是以他的实现分布式限流方案讲解一下，以代码的形式，顺便讲讲如何使用，并且简单原理实现。

1.相关依赖

org.redisson

redisson

3.11.4

2.限流器(RateLimiter)谷歌的guava实现的，也很强大

但是今天的主要的是

RRateLimiter

一个演化版的限流器，是的，多了一个R，它支持在分布式环境下，现在调用方的请求频率，可以实现不同Redisson实例下的多线程限流，也适用于相同实例Redisson下的多线程限流。是非公平性阻塞。

提供一下RRateLimiter 的接口文档

我们可以发现，基本上就是四个方法名，其余是重载

先分析一下第一个方法 acquire()

从此RateLimiter处获取许可，直到获得一个许可为止都将阻塞。

可以知道，是一个阻塞限流，直到获取到令牌。那我们可以对其分析一下源代码

可以看到基本上是给予了一个默认值1，一个许可证的数量

那么从public RFutrueacquireAsync(long permits)；分析一下

(1)先创建一个异步计算对象promise

(2)再调用方法tryAcquireAsync(permits, -1, null);

(3)然后将其结果，通过RFutrue返回

那么顺藤摸瓜，我们再分析一下方法：

public RFuturetryAcquireAsync(long permits, long timeout, TimeUnit unit) ；

如果有兴趣的童鞋可以看看源码，该对象，基本上所有的方法都是最终指向了上方法，只是参数，我们都封装好了。便于直接调用。

分析一下上述方法，主要做了一个什么事情呢？

就是将我们设定的超时时间统一转换为毫秒值，如果是-1，则不转换，直接为-1.

然后再定义一个异步任务，传递到

tryAcquireAsync(permits, promise, timeoutInMillis);

接下来就是重头戏了

一个私有方法：

(1)先记录进入该方法的起始时间 s

(2)进入下方法(执行一个lua脚本)

可想而知，这个就是redis获取令牌的命令，其中会判断是否超出获取许可数量。然后将其获取结果放回。返回的结果就是一个

Long类型数据，那么我们再通过异步计算，判断其返回结果是否为成功？是否等于NULL？

(3)判断e是否不等于null

如果不等于null，则代表获取到许可证，则立即返回

并将结果放在promise中，

如果等于null，则继续下面流程

(4)判断delay是否等于null

判断是否延迟，如果delay等于null，则将其异步标记为成功，也立即返回

(5)判断超时时间是否为-1

如果为-1，则进入递归任务，再获取一次许可，直至退出递归

(6)如果不等于-1，那说明存在确切的超时时间，那么将判断当前消耗的时间是否大于当前任务设定的超时时间

如果已经大于，则将立即返回，并标记结果失败

(7)再判断当前剩余的超时时间，是否小于延迟时间(delay)

(7.1)小于：则结束，并标记失败

(7.2)大于：则判断当前过去时间是否小于等于剩余超时时间，如果小于等于，则返回，标记失败，否则，则进入下个递归，并且传入剩余超时时间为最新超时时间。

3.如何实现分布式限流？

上代码！！

官方文档

基于Redis的分布式限流器(RateLimiter)可以用来在分布式环境下现在请求方的调用频率。既适用于不同Redisson实例下的多线程限流，也适用于相同Redisson实例下的多线程限流。该算法不保证公平性。除了同步接口外，还提供了异步(Async)、反射式(Reactive)和RxJava2标准的接口。

RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter");

// 初始化

// 最大流速 = 每1秒钟产生10个令牌

rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);

CountDownLatch latch = new CountDownLatch(2);

limiter.acquire(3);

// ...

Thread t = new Thread(() -> {

limiter.acquire(2);

// ...

});

aop实现

(1)监听注解

package cn.changemax.config.annotation;

import cn.changemax.enums.LimitTypeEnum;

import org.redisson.api.RateType;

import java.lang.annotation.ElementType;

import java.lang.annotation.Retention;

import java.lang.annotation.RetentionPolicy;

import java.lang.annotation.Target;

/**

* @author WangJi

* @Description cm限流注解

* @Date 2020/7/29 11:34

*/

@Target(ElementType.METHOD)

//方法上声明

@Retention(RetentionPolicy.RUNTIME)

public @interface CmLimit {

/**

* 资源名称，用于描述接口功能

*/

String name() default "";

/**

* 限制访问次数(单位时间内产生的令牌数)

*/

int count();

/**

* 时间间隔，单位秒

*/

int period();

/**

* 资源 key

*/

String key() default "";

/**

* 限制类型(ip/方法名)

*/

LimitTypeEnum limitType() default LimitTypeEnum.CUSTOMER;

/**

* RRateLimiter 速度类型

* OVERALL, //所有客户端加总限流

* PER_CLIENT; //每个客户端单独计算流量

* @return

*/

RateType mode() default RateType.PER_CLIENT;

}

(2)切点实现

package cn.changemax.config.aop;

import cn.changemax.commons.base.BaseAspectSupport;

import cn.changemax.config.annotation.CmLimit;

import cn.changemax.exception.ChangeMaxException;

import cn.changemax.utils.HttpContextUtil;

import cn.changemax.utils.IpInfoUtil;

import cn.changemax.utils.StringUtils;

import lombok.RequiredArgsConstructor;

import lombok.extern.slf4j.Slf4j;

import org.aspectj.lang.ProceedingJoinPoint;

import org.aspectj.lang.annotation.Around;

import org.aspectj.lang.annotation.Aspect;

import org.aspectj.lang.annotation.Pointcut;

import org.redisson.api.RRateLimiter;

import org.redisson.api.RateIntervalUnit;

import org.redisson.api.RedissonClient;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;

import java.lang.reflect.Method;

import java.util.concurrent.TimeUnit;

/**

* @author WangJi

* @Description redsson分布式限流器

*

* 技术参考文档：

* RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter");

* // 初始化

* // 最大流速 = 每10秒钟产生1个令牌

* rateLimiter.trySetRate(RateType.OVERALL, 1, 10, RateIntervalUnit.SECONDS);

* //需要1个令牌

* if(rateLimiter.tryAcquire(1)){

* //TODO:Do something

* }

*

*

* 高并发系统三把利器用于保护系统：缓存、降级和限流

* *缓存：缓存的目的就是提升系统的访问速度和增大系统处理容量

* *降级：降级是当服务出现问题或者影响到核心流程的时候，需要暂时屏蔽掉，待高峰或者问题解决后再打开

* *限流：限流的目的是通过对并发访问、请求进行限速，或者对一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务、排队或者等待、降级等处理方案。

*

* @Date 2020/7/29 11:00

*/

@Slf4j

@Aspect

@Component

@RequiredArgsConstructor

public class CmLimitAspect extends BaseAspectSupport {

private static final String CM_LIMIT_KEY_HEAD = "limit";

@Autowired

private RedissonClient redisson;

@Pointcut("@annotation(cn.changemax.config.annotation.CmLimit)")

public void pointcut() {

}

@Around("pointcut()")

public Object around(ProceedingJoinPoint point) throws Throwable {

HttpServletRequest request = HttpContextUtil.getHttpServletRequest();

Method method = resolveMethod(point);

CmLimit limit = method.getAnnotation(CmLimit.class);

String ip = IpInfoUtil.getIpAddr(request);

String key;

switch (limit.limitType()) {

case IP:

// ip类型

key = ip;

break;

case CUSTOMER:

//传统类型，采用注解提供key

key = limit.key();

break;

default:

//默认采用方法名

key = StringUtils.upperCase(method.getName());

}

// ImmutableListkeys = ImmutableList.of(StringUtils.join(CM_LIMIT_KEY_HEAD, limit.prefix(), ":", ip, key));

//生成key

final String ofRateLimiter = StringUtils.generateRedisKey(CM_LIMIT_KEY_HEAD, ip, key);

RRateLimiter rateLimiter = redisson.getRateLimiter(ofRateLimiter);

//设置访问速率，var2为访问数，var4为单位时间，var6为时间单

//每10秒产生1个令牌 总体限流

//创建令牌桶数据模型

rateLimiter.trySetRate(limit.mode(), limit.count(), limit.period(), RateIntervalUnit.SECONDS);

// permits 允许获得的许可数量 (如果获取失败，返回false) 1秒内不能获取到1个令牌，则返回，不阻塞

// 尝试访问数据，占数据计算值var1，设置等待时间var3

// acquire() 默认如下参数 如果超时时间为-1，则永不超时，则将线程阻塞，直至令牌补充

// 此处采用3秒超时方式，服务降级

if (!rateLimiter.tryAcquire(1, 2, TimeUnit.SECONDS)) {

log.error("IP【{}】访问接口【{}】超出频率限制，限制规则为[限流模式：{}; 限流数量：{}; 限流时间间隔：{};]",

ip, method.getName(), limit.mode().toString(), limit.count(), limit.period());

throw new ChangeMaxException("接口访问超出频率限制，请稍后重试");

}

return point.proceed();

}

}

(3)限流类型

package cn.changemax.enums;

/**

* @author WangJi

* @Description 限流类型

* @Date 2020/6/11 17:43

*/

public enum LimitTypeEnum {

/**

* 传统类型

*/

CUSTOMER,

/**

* 根据 IP地址限制

*/

IP

}

整个流程走完了，其中有个附加的知识点，了解一下：

CommandAsyncExecutor commandExecutor

RPromisepromise = new RedissonPromise();

分析第二个对象，我们可以看到上边源码中，都是返回成功，并且结果要么就是true，要么就是false，是因为我们再上层声明了结果为Boolean，可以理解为带有返回值的异步任务。

分析一下：Interface RPromise

所有的方法如下

刚刚源码中一直用到trySuccess，此文，我们了解一下这个方法即可

true当且仅当成功将这一未来标记为成功。否则，false因为此未来已被标记为成功或失败。并通知所有监听者

那么就说到这里，如果全文有什么错误的地方，积极欢迎各位指出错误，帮助大家一起成长，谢谢。