redis分布式限流的情况下，如果不能丢弃请求，那么被限流的请求如何处理？

落日映苍穹つ 2021-12-04 09:47 413阅读 0赞

**问题思考？**

*  redis分布式限流的情况下，如果请求不能被丢弃，那么被限流的请求该怎么处理呢？
 *  在大流量请求的情况下，如何保证接口的稳定性？

网络上大部分讲关于分布式限流，都是用redis+lua来实现，被限流的请求都是直接丢弃，然后在前端给用户提示友好信息。

这种做法当然没错，但是如果有一种业务场景不能丢弃请求，每个请求都有和业务相关的关键信息该怎么办呢？其实这时候就需要利用**kafka+aop**来实现。

**项目背景：**

*  需要同步第三方的数据到本地，第三方的接口做了限制，接口只能承受最多100次/秒的请求。   
 *  按日期的维度去同步90天的数据。

**实现方案：**

*  首先，在调用第三方接口的service上加上限流的注解，防止超频调用第三方接口而导致被加入黑名单。
 *  其次，利用redis+lua来做分布式限流。
 *  最后，利用aop拦截每一个调用服务的请求，验证此时的请求数量是否已经大于限制的数量，如果大于则将该请求放到kafak队列，进行下一次的处理。

**代码实现：**

*  自定义限流注解，加在需要被限制的请求方法上面。

/**
     * @Author: guandezhi
     * @Date: 2019/6/21 0:03
     */
    @Target({ElementType.METHOD})
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    public @interface DerbyRateLimiter {
    
        int limit() default 100; //限定次数
    
        int timeout() default 1; //超时时间
    
        int totalNum() default 0;
    
        String bizRemark() default "";
    }

*  redis+lua分布式限流实现

/**
     * @Author: guandezhi
     * @Date: 2019/6/21 0:03
     */
    public class CountRateLimiter {
    
        /**
         * @param jedis
         * @param limit   阈值/单位时间
         * @param seconds 单位时间 秒
         * @param name    名称
         * @return
         * @throws IOException
         */
        public static boolean acquire(
                Jedis jedis, int limit, int seconds, String name) throws IOException {
            String luaScript = "local key = KEYS[1] "
                    + "local limit = tonumber(ARGV[1]) "
                    + "local timeout = tostring(ARGV[2]) "
                    + "local current = tonumber(redis.call(\"INCRBY\", key,\"1\")) "
                    + "if current <= 1 then "
                    + "   redis.call(\"expire\", key,timeout) "
                    + "end "
                    + "if current  > limit then "
                    + "   return 0 "
                    + "else "
                    + "   return 1 "
                    + "end";
            return (Long) jedis.eval(luaScript, Lists.newArrayList(name), Lists.newArrayList(String.valueOf(limit), String.valueOf(seconds))) == 1;
        }
    }

*  利用aop实现请求拦截

/**
     * @Author: guandezhi
     * @Date: 2019/6/21 0:03
     */
    @Slf4j
    @Aspect
    @Component
    public class DerbyDynamicLimitAop {
    
        @Autowired
        private RedisTemplate redisTemplate;
        @Autowired
        private EmailService emailService;
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @Around("@annotation(rateLimiter)")
        public Object execute(ProceedingJoinPoint joinPoint, DerbyRateLimiter rateLimiter) {
            String methodName = joinPoint.getSignature().getName();
            int limit = rateLimiter.limit();
            int timeout = rateLimiter.timeout();
            String name = "travel:hotel:data:ratelimit:" + methodName + ":";
            Jedis jedis = JedisUtil.getJedis();
            Boolean success = null;
            long ttl = 0L;
            try {
                success = CountRateLimiter.acquire(jedis, limit, timeout, name);
                ttl = jedis.ttl(name) > 0 ? ttl : 0;
            } catch (Exception e) {
                log.error("DynamicLimitAop redis分布式限流异常：{}", e.getMessage());
            } finally {
                jedis.close();
            }
    
            try {
                if (!success) {
                    log.warn("动态拉取数据接口超频：当前频次：{}，限制频次：{}，休眠：{}秒", jedis.get(name), limit, ttl);
                    //将被限流的请求发送到队列中后续处理
                    kafkaTemplate.send(KafkaQueueTopic.KAFKA_QUEUE_DERBYDYNAMICLIMIT_LIST, getRequestParamsJson(joinPoint));
                } else {
                    return joinPoint.proceed();
                }
            } catch (Throwable e) {
                log.error("DynamicLimitAop异常：{}", e);
            }
            return new ResultVo<>();

注意：

1.此处的redis连接需要及时关闭，否则可能会导致redis连接数不够用的情况。

*  将队列中被限流的请求重新调用接口

/**
     * 监听被限流的请求
     *
     * @Author: guandezhi
     * @Date: 2019/5/14 15:06
     */
    @Slf4j
    @Component
    public class DerbyLimitAopKafkaListener {
    
        @Autowired
        @Qualifier(value = "derbyRoomService")
        private RoomService derbyRoomService;
    
        @Resource
        private ThreadPoolTaskExecutor taskExecutor;
    
    
        @KafkaListener(topics = KafkaQueueTopic.KAFKA_QUEUE_DERBYDYNAMICLIMIT_LIST, containerFactory = "batchFactory")
        public void listen(List<ConsumerRecord<String, String>> records) {
            try {
                if (CollectionUtils.isNotEmpty(records)) {
                    for (ConsumerRecord<String, String> record : records) {
                        RequestParam requestParam = JSONObject.parseObject(record.value(), RequestParam.class);
                        if (requestParam == null) {
                            continue;
                        }
                        log.info("开始批处理被限流的请求：{}", JSONObject.toJSONString(requestParam));
                        taskExecutor.execute(() -> {
                            try {
                                derbyRoomService.loadRoom(requestParam.getRequest(), requestParam.getData());
                            } catch (Exception e) {
                                log.error("DerbyLimitAopKafkaListener, requestParam {},异常 {}", JSONObject.toJSONString(requestParam), e.toString());
                            }
                        });
                    }
                }
            } catch (Exception e) {
                log.error("DerbyLimitAopKafkaListener批量消费异常 records：{},异常信息 {}",
                        JSONObject.toJSONString(records), e.getMessage());
            }
        }
    }

这样就实现了被限流的请求重新进入接口调用的循环中。虽然有被限流，但是请求都没有被丢弃。