SnowJean使用以及原理（SpringBoot）

灰太狼 2023-02-14 01:38 1阅读 0赞

# 一.使用 #

## 1.POM文件 ##

<dependency>
                <groupId>cn.yueshutong</groupId>
                <artifactId>snowjean-spring-boot-starter</artifactId>
                <version>3.0.0.RELEASE</version>
            </dependency>

## 2.配置 ##

import cn.yueshutong.commoon.entity.RateLimiterRule;
    import cn.yueshutong.commoon.entity.RateLimiterRuleBuilder;
    import cn.yueshutong.core.config.RateLimiterConfig;
    import cn.yueshutong.core.limiter.RateLimiter;
    import cn.yueshutong.core.limiter.RateLimiterFactory;
    import cn.yueshutong.core.observer.RateLimiterObserver;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @Classname RateLimiterConfig
     * @Description TODO
     * @Date 2020/5/27 13:41
     * @Created zzf
     */
    @Configuration
    public class RateLimiterRuleConfig {
    
        @PostConstruct
        public void register() {
            RateLimiterRule rateLimiterRule = new RateLimiterRuleBuilder()
                    //ID很重要，对应注解@Limiter中的value
                    .setId("limiter")
                    .setLimit(1)
                    .setBatch(1)
    //                .setLimiterModel(LimiterModel.CLOUD)
                    .build();
            // 2.配置TicketServer地址（支持集群、加权重）
            Map<String, Integer> map = new HashMap<>();
            map.put("127.0.0.1:8521", 1);
            // 3.全局配置
            RateLimiterConfig config = RateLimiterConfig.getInstance();
            config.setTicketServer(map);
            //生产限流器
            RateLimiter rateLimiter = RateLimiterFactory.of(rateLimiterRule, config);
            //随时随地获取已生产的限流器
            RateLimiter rateLimiter1 = RateLimiterObserver.getMap().get(rateLimiter.getId());
        }
    }

## 3.使用 ##

@RestController
    @RequestMapping("/api/v3/")
    @Api(value = "open_api", description = "open_api基础接口", tags = {"open_api"})
    public class TestController {
    
        @ApiOperation(value = "测试", notes = "测试")
        @ApiResponses({
                @ApiResponse(code = 400, message = "参数非法"),
                @ApiResponse(code = 500, message = "服务器错误"),
                @ApiResponse(code = 200, message = "成功")
        })
        @ApiImplicitParams({
        })
        @Limiter(value = "limiter")
        @RequestMapping(value = "/test", method = RequestMethod.GET)
        public CommonResult<String> getUserInfo() {
            return new CommonResult<>("999");
        }
    }

## 4.测试 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1][]

**当你在短时间内快速调用时就会进行限制！**

# **二.原理** #

**这里我只分析****本地限流：**

**首先它会通过在方法上的注解@Limiter对调用方法进行拦截：**

package cn.yueshutong.annotation.aspect;
    
    import cn.yueshutong.annotation.entity.Limiter;
    import cn.yueshutong.core.observer.RateLimiterObserver;
    import org.aspectj.lang.ProceedingJoinPoint;
    import org.aspectj.lang.Signature;
    import org.aspectj.lang.annotation.Around;
    import org.aspectj.lang.annotation.Aspect;
    import org.aspectj.lang.annotation.Pointcut;
    import org.aspectj.lang.reflect.MethodSignature;
    
    import java.lang.reflect.Method;
    
    @Aspect
    public class RateLimiterAspect {
    
        @Pointcut("@annotation(cn.yueshutong.annotation.entity.Limiter)")
        public void pointcut() {
        }
    
        @Around("pointcut() && @annotation(limiter)")
        public Object around(ProceedingJoinPoint pjp, Limiter limiter) throws Throwable {
            cn.yueshutong.core.limiter.RateLimiter rateLimiter = RateLimiterObserver.getMap().get(limiter.value());
            if (rateLimiter.tryAcquire()) {
                return pjp.proceed();
            }
            Signature sig = pjp.getSignature();
            if (!(sig instanceof MethodSignature)) {
                throw new IllegalArgumentException("This annotation can only be used in methods.");
            }
            MethodSignature msg = (MethodSignature) sig;
            Object target = pjp.getTarget();
            Method fallback = target.getClass().getMethod(limiter.fallback(), msg.getParameterTypes());
            return fallback.invoke(target,pjp.getArgs());
        }
    
    }

**首先根据limiter的value获取相应的限流器，然后通过调用内部的tryAcquire方法判断是否通过，如果是true则通过pjp.proceed()进入下一个流程，否则他会获取@Limiter中的fallback方法进行回调。**

**下面来看看它的具体实现逻辑：**

/**
         * 2.Check
         */
        @Override
        public boolean tryAcquire() {
            if (rule.isEnable()) {
                //限流功能已关闭
                return true;
            }
            return tryAcquireMonitor();
        }
    
    
        /**
         * 3.Monitor
         */
        private boolean tryAcquireMonitor() {
            if (rule.getLimiterModel() == LimiterModel.POINT) {
                //本地限流不支持监控
                return tryAcquirePut();
            }
            MonitorBean monitor = new MonitorBean();
            monitor.setLocalDateTime(LocalDateTime.now());
            monitor.setPre(1);
            monitor.setApp(rule.getApp());
            monitor.setId(rule.getId());
            monitor.setName(rule.getName());
            monitor.setMonitor(rule.getMonitor());
            boolean b = tryAcquirePut(); //fact
            if (b) {
                monitor.setAfter(1);
            }
            config.getScheduledThreadExecutor().execute(() -> { //异步执行
                monitorService.save(monitor);
            });
            return b;
        }
    
    
        /**
         * 4.putCloudBucket
         */
        private boolean tryAcquirePut() {
            boolean result = tryAcquireFact();
            //分布式方式下检查剩余令牌数
            putCloudBucket();
            return result;
        }

进入**tryAcquire方法后他会判断是否开启限流功能，然后通过tryAcquireMonitor方法进行调用。通过调用可以看到不管是分布式还是本地，它内部的核心都是tryAcquireFact方法！**

private final AtomicLong bucket = new AtomicLong(0); //令牌桶初始容量：0
    
        /**
         * 5.tryAcquireFact
         */
        private boolean tryAcquireFact() {
            if (rule.getLimit() == 0) {
                return false;
            }
            boolean result = false;
            switch (rule.getAcquireModel()) {
                case FAILFAST:
                    result = tryAcquireFailed();
                    break;
                case BLOCKING:
                    result = tryAcquireSucceed();
                    break;
            }
            return result;
        }
    
    
        /**
         * CAS获取令牌,没有令牌立即失败
         */
        private boolean tryAcquireFailed() {
            long l = bucket.longValue();
            while (l > 0) {
                if (bucket.compareAndSet(l, l - 1)) {
                    return true;
                }
                l = bucket.longValue();
            }
            return false;
        }
    
    
        /**
         * CAS获取令牌,阻塞直到成功
         */
        private boolean tryAcquireSucceed() {
            long l = bucket.longValue();
            while (!(l > 0 && bucket.compareAndSet(l, l - 1))) {
                sleep();
                l = bucket.longValue();
            }
            return true;
        }

**这里可以看到其内部是通过对一个原子对象的cas操作，如果backet值大于0且cas操作成功则返回true，否则就是false。这就是其内部取令牌的操作。如果是分布式限流话它还有一个putCloudBucket方法对总数进行一个同步：**

/**
         * 集群限流，取批令牌
         */
        private void putCloudBucket() {
            //校验
            if (!rule.getLimiterModel().equals(LimiterModel.CLOUD) ||
                    bucket.get() / 1.0 * rule.getBatch() > rule.getRemaining()) {
                return;
            }
            //异步任务
            config.getScheduledThreadExecutor().execute(() -> {
                //DCL,再次校验
                if (bucket.get() / 1.0 * rule.getBatch() <= rule.getRemaining()) {
                    synchronized (bucket) {
                        if (bucket.get() / 1.0 * rule.getBatch() <= rule.getRemaining()) {
                            String result = config.getTicketServer().connect(RateLimiterConfig.http_token, JSON.toJSONString(rule));
                            if (result != null) {
                                bucket.getAndAdd(Long.parseLong(result));
                            }
                        }
                    }
                }
            });
        }
    
    
    
        public String connect(String path, String data) {
            String server = getServer();
            try {
                return HttpUtil.connect("http://" + server + "/" + path)
                        .setData("data", data)
                        .setMethod("POST")
                        .execute()
                        .getBody();
            } catch (IOException e) {
                if (System.currentTimeMillis() - start >3000) {
                    logger.error("{} The server is not available.", server);
                    start = System.currentTimeMillis();
                }
                serverList.remove(server);
                backupsList.add(server);
            }
            return null;
        }

**他会启动一个定时任务，通过远程调用的方式对令牌进行一个同步操作！所以其实SnowJean的原理很简单。**

**最后看看生成令牌的操作：**

/**
         * 本地限流，放入令牌
         */
        private void putPointBucket() {
            if (this.scheduledFuture != null) {
                this.scheduledFuture.cancel(true);
            }
            if (rule.getLimit() == 0 || !rule.getLimiterModel().equals(LimiterModel.POINT)) {
                return;
            }
            this.scheduledFuture = config.getScheduledThreadExecutor().scheduleAtFixedRate(() -> bucket.set(rule.getLimit()), rule.getInitialDelay(), rule.getPeriod(), rule.getUnit());
        }

**如果是本地限流的话它其实是通过一个定时任务去不断生成的！**

**文档：**[https://github.com/ystcode/SnowJena/blob/master/CN\_README.md][https_github.com_ystcode_SnowJena_blob_master_CN_README.md]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200530145700687.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200530145709255.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx,size_16,color_FFFFFF,t_70
[https_github.com_ystcode_SnowJena_blob_master_CN_README.md]: https://github.com/ystcode/SnowJena/blob/master/CN_README.md