今天来讲讲分布式环境下，怎么达到对象共享，以及实现原子性（atomic），以Redis中的Redisson为例（实现分布式锁、分布式限流等）

今天药忘吃喽~ 2021-10-09 08:52 267阅读 0赞

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

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

![20200729141227245.png][]

关于解决对象共享问题，很多方式，通过一般的关系型数据库就可以（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官方文档][Redisson]**

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

![20200729142658917.png][]

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

![20200729142744937.png][]

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

1.相关依赖

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.11.4</version> 
    </dependency>

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

但是今天的主要的是

RRateLimiter

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

提供一下[RRateLimiter 的接口文档][RRateLimiter]

![20200729143722643.png][]

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

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

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

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

![20200730082436629.png][]

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

那么从public RFutrue<Void> acquireAsync(long permits)；分析一下

（1）先创建一个异步计算对象promise

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

（3）然后将其结果，通过RFutrue返回

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

public RFuture<Boolean> tryAcquireAsync(long permits, long timeout, TimeUnit unit) ；

![20200730083128296.png][]

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

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

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

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

tryAcquireAsync(permits, promise, timeoutInMillis);

接下来就是重头戏了

![20200730083605640.png][]

一个私有方法：

（1）先记录进入该方法的起始时间 s

（2）进入下方法（执行一个lua脚本）

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

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

![20200730083845173.png][]

（3）判断e是否不等于null

![20200730084442970.png][]

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

并将结果放在promise中，

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

（4）判断delay是否等于null

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

![20200730085147255.png][]

（5）判断超时时间是否为-1

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

![20200730085138812.png][]

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

![20200730085302961.png][]

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

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

（7.1）小于：则结束，并标记失败

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

![20200730090246318.png][]

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());
            }
    //        ImmutableList<String> keys = 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

RPromise<Boolean> promise = new RedissonPromise<Boolean>();

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

分析一下：Interface RPromise<T>

![20200730090907842.png][]

所有的方法如下

![20200730090948727.png][]

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

![20200730091047745.png][]

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

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

[20200729141227245.png]: /images/20211009/3a465fcabf6d48dd8fb792ea6fe8a0f9.png
[Redisson]: https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
[20200729142658917.png]: /images/20211009/38e12dc96fd9475d80205a9e958591e0.png
[20200729142744937.png]: /images/20211009/2f18ed7b5fa74af38290ec65b171c162.png
[RRateLimiter]: https://www.javadoc.io/doc/org.redisson/redisson/3.10.6/org/redisson/api/RRateLimiter.html
[20200729143722643.png]: /images/20211009/83cfe2859da648dca373e362dc8eaf1f.png
[20200730082436629.png]: /images/20211009/f4314a54d893488caf9c0f06dbb7f606.png
[20200730083128296.png]: /images/20211009/7f6c78b45fa145efb5b19b0cc5d56a2f.png
[20200730083605640.png]: /images/20211009/f38e5d95f47a4a68b39584b933562bbf.png
[20200730083845173.png]: /images/20211009/5883cf27134a416784160521ccfcc29d.png
[20200730084442970.png]: /images/20211009/38255aa2218d4ecd963ee3789f40385f.png
[20200730085147255.png]: /images/20211009/44f0d182eb3a4a1286bc06ccf9bd3694.png
[20200730085138812.png]: /images/20211009/0624a45fa0004a86a66288c5a4ea6dc8.png
[20200730085302961.png]: /images/20211009/60ea12c5525f4dda834b1ad11694d974.png
[20200730090246318.png]: /images/20211009/ee9865e5323c44da903ab99c12611102.png
[20200730090907842.png]: /images/20211009/b516836b6789487eb09c4dd0802440e9.png
[20200730090948727.png]: /images/20211009/666789a801c74d84a178e4a156e2b052.png
[20200730091047745.png]: /images/20211009/242bf69b3a3b4aa68767f39e6dc39f35.png