从0到1构造自定义限流组件 | 京东云技术团队-低调大师

从0到1构造自定义限流组件 | 京东云技术团队

2023-06-20 770

一背景

在系统高可用设计中，接口限流是一个非常重要环节，一方面是出于对自身服务器资源的保护，另一方面也是对依赖资源的一种保护措施。比如对于 Web 应用，我限制单机只能处理每秒 1000 次的请求，超过的部分直接返回错误给客户端。虽然这种做法损害了用户的使用体验，但是它是在极端并发下的无奈之举，是短暂的行为，因此是可以接受的。

二设计思路

常见的限流有2种思路

第一种是限制总量,也就是限制某个指标的累积上限，常见的是限制当前系统服务的用户总量，例如：某个抢购活动商品数量只有 100 个，限制参与抢购的用户上限为 1 万个，1 万以后的用户直接拒绝。
第二种是限制时间量，也就是限制一段时间内某个指标的上限，例如 1 分钟内只允许 10000 个用户访问；每秒请求峰值最高为 10 万。

三限流算法

目前实现限流算法主要分为3类，这里不详细展开介绍：

1）时间窗口

固定时间窗口算法是最简单的限流算法，它的实现原理就是控制单位时间内请求的数量，但是这个算法有个缺点就是临界值问题。
为了解决临界值的问题，又推出滑动时间窗口算法，其实现原理大致上是将时间分为一个一个小格子，在统计请求数量的时候，是通过统计滑动时间周期内的请求数量。

2）漏斗算法

漏斗算法的核心是控制总量，请求流入的速率不确定，超过流量部分益出，该算法比较适用于针对突发流量，想要尽可能的接收全部请求的场景。其缺点也比较明显，这个总量怎么评估，大小怎么配置，而且一旦初始化也没法动态调整。

3）令牌桶算法

令牌桶算法的核心是控制速率，令牌产生的速度是关键，不断的请求获取令牌，获取不到就丢弃。该算法比较适用于针对突发流量，以保护自身服务资源以及依赖资源为主，支持动态调整速率。缺点的话实现比较复杂，而且会丢弃很多请求。

四实现步骤

我们自定义的这套限流组件有是基于guava RateLimiter封装的，采用令牌桶算法以控制速率为主，支持DUCC动态配置，同时支持限流后的降级措施。接下来看一下整体实现方案

1、自定义RateLimiter Annotation标签

这里主要对限流相关属性的一个定义，包括每秒产生的令牌数、获取令牌超时时间、降级逻辑实现以及限流开关等内容

@Documented
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface SysRateLimit {

    /**
     * 每秒产生的令牌数 默认500
     *
     * @return
     */
    double permitsPerSecond() default 500D;

    /**
     * 获取令牌超时时间 默认100
     *
     * @return
     */
    long timeout() default 100;

    /**
     * 获取令牌超时时间单位 默认毫秒
     *
     * @return
     */
    TimeUnit timeUnit() default TimeUnit.MILLISECONDS;

    /**
     * 服务降级方法名称 Spring bean id
     *
     * @return
     */
    String fallbackBeanId() default "";

    /**
     * 限流key 唯一
     *
     * @return
     */
    String limitKey() default "";
}

2、基于Spring Aspect 构造切面

首先就是我们需要构造一个Aspect切面用于扫描我们自定义的SysRateLimit标签

@Slf4j
@EnableAspectJAutoProxy
@Aspect
public class SysRateLimitAspect {
    
    /**
     * 自定义切入点
     */
    @Pointcut("@annotation(com.jd.smb.service.ratelimiter.annotation.SysRateLimit)")
    public void pointCut() {

    }

    /**
     * 方法前执行限流方案
     *
     * @param joinPoint
     * @return
     * @throws Throwable
     */
    @Around("pointCut()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        // 如果未获取到对象，直接执行方法
        if (signature == null) {
            return joinPoint.proceed();
        }

        try {
            Method method = joinPoint.getTarget().getClass().getDeclaredMethod(signature.getName(), signature.getMethod().getParameterTypes());
            // 获取注解对象
            SysRateLimit sysRateLimit = method.getAnnotation(SysRateLimit.class);
            if (sysRateLimit == null) {
                return joinPoint.proceed();
            }
            
        } catch (Exception e) {
            // todo log
        }
        return joinPoint.proceed();
    }
}

获取自定义SysRateLimit标签的各种属性

 // 限流key
String limitKey = sysRateLimit.limitKey();
if (StringUtils.isBlank(limitKey)) {
    return joinPoint.proceed();
}
// 令牌桶数量
double permitsPerSecond = sysRateLimit.permitsPerSecond();
// 获取令牌超时时间
long timeout = sysRateLimit.timeout();
// 获取令牌超时时间单位
TimeUnit timeUnit = sysRateLimit.timeUnit();

将我们自定义的SysRateLimiter 和 Guava RateLimiter 进行整合

首先我们需要构造一个全局Map，用于存储我们开启限流的方法，key就是我们定义的limitKey， value就是我们转换后的Guava RateLimiter

 /**
 * 存储RateLimiter（key: limitKey value:RateLimiter ）
 */
private static final Map<String, RateLimiter> LIMITER_MAP = new ConcurrentHashMap<>();

接着就是核心逻辑：这里首先从我们创建的Map中获取Guava RateLimiter，获取不到就创建RateLimiter.create(permitsPerSecond) ；然后调用RateLimiter.tryAcquire()尝试获取令牌桶，获取成功则执行后续的逻辑，这里重点获取失败后，我们需要执行我们的降级方法。(注意：Guava RateLimiter 有很多API，这里我们不展开讨论，后续会针对Guava限流的源码进行详细的解析)

RateLimiter rateLimiter;
// Map中是否存在 存在直接获取
if (LIMITER_MAP.containsKey(limitKey)) {
    rateLimiter = LIMITER_MAP.get(limitKey);
} else {
    // 不存在创建后放到Map中
    rateLimiter = RateLimiter.create(permitsPerSecond);
    LIMITER_MAP.put(limitKey, rateLimiter);
}
// 尝试获取令牌
if (!rateLimiter.tryAcquire(timeout, timeUnit)) {
    // todo 限流后降级措施
    return this.fallBack(sysRateLimit, joinPoint, signature);
}

降级方案执行

上面我们在获取令牌桶超时后，需要执行我们的降级逻辑，怎么做呢？也很简单，我们在定义SysRateLimiter的时候有个fallBackBeanId，这个就是我们执行降级逻辑的bean对象Id，需要我们提前进行创建。接着我们看一下是怎么实现的。

    /**
     * 执行降级逻辑
     *
     * @param sysRateLimit
     * @param joinPoint
     * @param signature
     * @return
     */
    private Object fallBack(SysRateLimit sysRateLimit, ProceedingJoinPoint joinPoint, MethodSignature signature) {
        String fallbackBeanId = sysRateLimit.fallbackBeanId();
        // 当没有配置具体的降级实现方案的时候 可以结合业务世纪情况设置限流错误码
        if (StringUtils.isBlank(fallbackBeanId)) {
            // 自定义的 可以结合自己系统里的进行设置
            return ApiResult.error(ResultCode.REACH_RATE_LIMIT);
        }

        try {
            // SpringContext中通过BeanId获取对象 SpringUtils只是获取bean对象的工具类 有多种实现方式 可自行百度
            Object bean = SpringUtils.getBean(fallbackBeanId);
            Method method = bean.getClass().getMethod(signature.getName(), signature.getParameterTypes());
            // 执行对应的方法
            return method.invoke(bean, joinPoint.getArgs());
        } catch (Exception e) {
            // todo error log
        }
        return ApiResult.error(ResultCode.REACH_RATE_LIMIT);
    }

这样我们大概的一个架子就弄好了。接下来我们看看实际该如何使用

3、具体应用

在方法入口引入SysRateLimiter标签

@Slf4j
@RestController
@RequestMapping("/api/user")
@RequiredArgsConstructor
public class UserQueryController extends AbstractController {

    /**
     * 查询用户信息
     *
     * @param request
     * @return
     */
    @GetMapping("/info/{id}")
    @SysRateLimit(permitsPerSecond = 500, limitKey = "UserQueryController.info", fallbackBeanId = "userQueryControllerFallBack",
            timeout = 100, timeUnit = TimeUnit.MILLISECONDS)
    public ApiResult<UserInfo> info(@PathVariable Long id, HttpServletRequest request) {
        // todo 业务逻辑查询 这里不展开
        return ApiResult.success();
    }
}

设置降级方法

@Service
public class UserQueryControllerFallBack {

    /**
     * 降级后执行的逻辑
     *
     * @param request
     * @return
     */
    public ApiResult<UserInfo> info(Long id, HttpServletRequest request) {
        // todo 编写限流降级后的逻辑 可以是降级码 也可以是默认对象
        return ApiResult.success(null);
    }
}

当请求进来的时候，会结合我们设置的阈值进行令牌桶的获取，获取失败后会执行限流，这里我们进行了限流后的降级处理。其实到这里我们完成限流组件的简单封装和使用，但是仍有一些点需要我们进行处理，例如如何动态设置令牌的数量，接下来我们就看一下如何实现令牌的动态设置。

4、动态设置令牌数量

通过DUCC配置令牌数量我们需要定义一个DUCC配置，这里面内容很简单，配置我们设置limitKey的令牌数量

@Data
@Slf4j
@Component
public class RateLimitConfig {

    /**
     * 配置config key: limitKey value: 数量
     */
    private Map<String, Integer> limitConfig;

    /**
     * 监听ducc配置
     *
     * @param json
     */
    @LafValue(key = "rate.limit.conf")
    public void setConfig(String json) {
        if (StringUtils.isBlank(json)) {
            return;
        }
        Map<String, Integer> map = JsonModelUtils.getModel(json, Map.class, null);
        if (map != null) {
            Wrapper.wrapperBean(map, this, true);
        }
    }
}

通过DUCC配置获取指定limitKey的令牌数量，获取失败则采用方法设置默认数量，这样我们后面设置令牌数量就可以通过DUCC动态的配置了

 /**
     * 获取令牌桶数量
     *
     * @param sysRateLimit
     * @return
     */
    private double getPermitsPerSecond(SysRateLimit sysRateLimit) {
        // 方法默认令牌数量
        double defaultValue = sysRateLimit.permitsPerSecond();
        if (rateLimitConfig == null || rateLimitConfig.getLimitConfig() == null) {
            return defaultValue;
        }
        // 配置的令牌数量
        Integer value = rateLimitConfig.getLimitConfig().get(sysRateLimit.limitKey());
        if (value == null) {
            return defaultValue;
        }
        return value;
    }

5、后续其他配置

其实后续我们的其他属性都可以通过DUCC动态化的来配置，这里呢因为和令牌桶数量类似，就不再展开描述了。感兴趣的小伙伴可以自行设置，根据我们的使用，使用默认配置即可。

作者：京东零售王磊

来源：京东云开发者社区

微信关注我们

原文链接：https://my.oschina.net/u/4090830/blog/10083862

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

规则引擎调研及初步使用 | 京东云技术团队

一、产生的背景生产过程中，线上的业务规则内嵌在系统的各处代码中，每次策略的调整都需要更新线上系统，进行从需求->设计->编码->测试->上线这种长周期的流程，满足不了业务规则的快速变化以及低成本的更新试错迭代。因此需要有一种解决方案将商业决策逻辑和应用开发者的技术决策分离开，在系统运行时能去更新管理业务规则。规则引擎(业务规则管理系统，英文名为BRMS（即Business Rule Management System））正是这样的解决方案。二、实际业务场景：一个小例子：假如我们有个业务场景，当客户的积分位于一个区间A时，我们给予他一个头衔a，当一个客户的积分位于区间B时，我们给予他一个头衔b，当客户的积分位于区间C时，我们给予他一个头衔c。如果我们使用if-else-then来写，是可以实现的，但是这里存在一个问题：规则定义和代码耦合在一起了。如果我们改变规则，把区间A，B，C 改成D，E，F，又或是将规则增加，从3组变为100组，那么我们改代码实在是太麻烦了。这时候规则引擎就派上用场了，我们希望把规则和代码解耦，形成一个规则引擎，以适应复杂多变的...

2023-06-20

861

作者 | 百度智能小程序团队导读本文首先介绍了分布式服务下日志服务建设的挑战，然后介绍了下业内ELK的通用解决方案及与天眼日志服务的差异性，接下来详细介绍了天眼日志服务平台的整体架构，如何做采集、传输、检索、隔离、清理等机制的，最后对日志服务与大模型进行结合，不断探索效能的提升。全文11796字，预计阅读时间30分钟。 01 分布式服务下日志服务挑战分布式服务系统中，每个服务有大量的服务器，而每台服务器每天都会产生大量的日志。我们面临的主要挑战有： 1、日志量巨大：在分布式服务环境中，日志分散在多个节点上，每个服务都会产生大量的日志，因此需要一种可靠的机制来收集和聚合日志数据。 2、多样化的日志格式：不同的服务可能使用不同的日志格式，例如日志输出的字段、顺序和级别等，这会增加日志服务的开发和维护难度。 3、日志服务的可扩展性和可靠性：随着分布式服务数量的增加和规模的扩大，日志服务需要能够进行横向扩展和纵向扩展，以保证其性能和可靠性。所以我们该如何提供分布式系统下高效、低延迟、高性能的日志服务呢？ 02 业内ELK通用解决方案 2.1 Elastic Stack发展历程 2.2...

2023-06-20

799

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。