设计模式之美-课程笔记51-项目实战1-设计实现一个支持各种算法的限流框架

Published on December 17, 2023

项目背景

一个金融创业项目早期搭建，随着发展出多个app，代码复制+修改出新app；
随着发展这种模式bug要修改就要修改所有的app，上了通用功能也需要一起所有人一起改；
而且历史代码存在臃肿可读性差，缺乏设计等问题；
为了提升开发效率一个方法就是抽离出通用模块集中维护，其他的app通过远程接口调用功能即可。易于维护，降低人工成本。

需求背景

因为调用方的不恰当使用、bug或者因为活动带来的突发流量，导致接口响应超过常规时间甚至超时。
所以想开发一个rate limiter。且希望这个项目能够带来尽可能大的影响在公司范围内（通用框架）-》抽象意识、框架意识。

需求分析

在配置文件设置限流规则，应用在启动时读取；

configs:
- appId: app-1
  limits:
  - api: /v1/user
    limit: 100
  - api: /v1/order
    limit: 50
- appId: app-2
  limits:
  - api: /v1/user
    limit: 50
  - api: /v1/order
    limit: 50

应用接收到请求后发送给限流框架处理继续还是熔断；

String appId = "app-1"; // 调用方APP-ID
String url = "http://www.eudemon.com/v1/user/12345";// 请求url
RateLimiter ratelimiter = new RateLimiter();
boolean passed = ratelimiter.limit(appId, url);
if (passed) {
  // 放行接口请求，继续后续的处理。
} else {
  // 接口请求被限流。
}

非功能性需求

易用性：规则的配置、编程接口的使用都很简单；易于集成的Spring（根据业务需求）。

扩展性、灵活性：灵活的扩展限流算法；支持不同格式的配置。

性能：低延迟-因为每个请求都会被检查。

容错性：限流服务的出错不能影响正常的应用功能请求-因为本身就是一个辅助增强的功能，应该是尽量减少他对原来应用的影响。

限流规则

语法格式：调用方、时间粒度、接口、阈值
时间粒度越细限流整形效果越好但是不是绝对的，要根据实际情况来决定；
配置方式

限流算法

有多种限流算法，为了提高易用性和扩展性，需要预留扩展点可以后期增加更多算法。默认实现中也可以提前实现常见算法供用户选择。

限流模式

单机: 不是指一台物理机而是一个服务实例。
分布式：多个实例的总访问频率限流。
主要区别在于接口访问计数器的实现：分布式的一般需要一个集中的计数器。
高容错、低延迟一般就是说的是基于Redis实现的分布式限流方案。
Redis的异常可以被包裹抛出但是超时难处理。一般是要设置超时时间，超时即判定为限流失效继续执行接口请求。

集成使用

低耦合
也可以参照MyBatis的类库思想（MyBatis-Spring类库方便Spring开发），设计一个RateLimiter-Spring类库。

V1版本

接口只支持HTTP接口（URL)，限流规则只支持本地YAML，算法只支持fixed-window。

最小原型代码

面向对象设计：划分职责识别类、定义属性和方法、定义类之间的交互、组装类并提供执行入口。

com.xzg.ratelimiter
  --RateLimiter
com.xzg.ratelimiter.rule
  --ApiLimit
  --RuleConfig
  --RateLimitRule
com.xzg.ratelimiter.alg
  --RateLimitAlg

RateLimiter类

串联：解析ruleConfig

执行limit

public class RateLimiter {
  private static final Logger log = LoggerFactory.getLogger(RateLimiter.class);
  // 为每个api在内存中存储限流计数器
  private ConcurrentHashMap<String, RateLimitAlg> counters = new ConcurrentHashMap<>();
  private RateLimitRule rule;

  public RateLimiter() {
    // 将限流规则配置文件ratelimiter-rule.yaml中的内容读取到RuleConfig中
    InputStream in = null;
    RuleConfig ruleConfig = null;
    try {
      in = this.getClass().getResourceAsStream("/ratelimiter-rule.yaml");
      if (in != null) {
        Yaml yaml = new Yaml();
        ruleConfig = yaml.loadAs(in, RuleConfig.class);
      }
    } finally {
      if (in != null) {
        try {
          in.close();
        } catch (IOException e) {
          log.error("close file error:{}", e);
        }
      }
    }

    // 将限流规则构建成支持快速查找的数据结构RateLimitRule
    this.rule = new RateLimitRule(ruleConfig);
  }

  public boolean limit(String appId, String url) throws InternalErrorException {
    ApiLimit apiLimit = rule.getLimit(appId, url);
    if (apiLimit == null) {
      return true;
    }

    // 获取api对应在内存中的限流计数器（rateLimitCounter）
    String counterKey = appId + ":" + apiLimit.getApi();
    RateLimitAlg rateLimitCounter = counters.get(counterKey);
    if (rateLimitCounter == null) {
      RateLimitAlg newRateLimitCounter = new RateLimitAlg(apiLimit.getLimit());
      rateLimitCounter = counters.putIfAbsent(counterKey, newRateLimitCounter);
      if (rateLimitCounter == null) {
        rateLimitCounter = newRateLimitCounter;
      }
    }

    // 判断是否限流
    return rateLimitCounter.tryAcquire();
  }
}

RuleConfi和ApiLimit类

从代码中，我们可以看出来，RuleConfig 类嵌套了另外两个类 AppRuleConfig 和 ApiLimit。这三个类跟配置文件的三层嵌套结构完全对应。我把对应关系标注在了下面的示例中，你可以对照着代码看下。

public class RuleConfig {
  private List<AppRuleConfig> configs;

  public List<AppRuleConfig> getConfigs() {
    return configs;
  }

  public void setConfigs(List<AppRuleConfig> configs) {
    this.configs = configs;
  }

  public static class AppRuleConfig {
    private String appId;
    private List<ApiLimit> limits;

    public AppRuleConfig() {}

    public AppRuleConfig(String appId, List<ApiLimit> limits) {
      this.appId = appId;
      this.limits = limits;
    }
    //...省略getter、setter方法...
  }
}

public class ApiLimit {
  private static final int DEFAULT_TIME_UNIT = 1; // 1 second
  private String api;
  private int limit;
  private int unit = DEFAULT_TIME_UNIT;

  public ApiLimit() {}

  public ApiLimit(String api, int limit) {
    this(api, limit, DEFAULT_TIME_UNIT);
  }

  public ApiLimit(String api, int limit, int unit) {
    this.api = api;
    this.limit = limit;
    this.unit = unit;
  }
  // ...省略getter、setter方法...
}

configs:          <!--对应RuleConfig-->
- appId: app-1    <!--对应AppRuleConfig-->
  limits:
  - api: /v1/user <!--对应ApiLimit-->
    limit: 100
    unit：60
  - api: /v1/order
    limit: 50
- appId: app-2
  limits:
  - api: /v1/user
    limit: 50
  - api: /v1/order
    limit: 50

RateLimitRule

限流过程会频繁查询限流规则，为了提升速度将限流规则组织成为一种支持按照URL快速查询的数据结构（Trie）。

public class RateLimitRule {
  public RateLimitRule(RuleConfig ruleConfig) {
    //...
  }

  public ApiLimit getLimit(String appId, String api) {
    //...
  }
}

RateLimitAlgo

这个类是限流算法实现类。它实现了最简单的固定时间窗口限流算法。每个接口都要在内存中对应一个 RateLimitAlg 对象，记录在当前时间窗口内已经被访问的次数。

Review 最小原型代码

可读性

模块划分相对合理

RateLimiter类主要做组装，最好不要有过多的业务逻辑。

扩展性

RateLimitAlg和RateLimitRule都没有基于接口去实现，可扩展性差一些；

RateLimiter中yaml都是硬编码。

重构最小原型代码

// 重构前：
com.xzg.ratelimiter
  --RateLimiter
com.xzg.ratelimiter.rule
  --ApiLimit
  --RuleConfig
  --RateLimitRule
com.xzg.ratelimiter.alg
  --RateLimitAlg
  
// 重构后：
com.xzg.ratelimiter
  --RateLimiter(有所修改)
com.xzg.ratelimiter.rule
  --ApiLimit(不变)
  --RuleConfig(不变)
  --RateLimitRule(抽象接口)
  --TrieRateLimitRule(实现类，就是重构前的RateLimitRule）
com.xzg.ratelimiter.rule.parser
  --RuleConfigParser(抽象接口)
  --YamlRuleConfigParser(Yaml格式配置文件解析类)
  --JsonRuleConfigParser(Json格式配置文件解析类)
com.xzg.ratelimiter.rule.datasource
  --RuleConfigSource(抽象接口)
  --FileRuleConfigSource(基于本地文件的配置类)
com.xzg.ratelimiter.alg
  --RateLimitAlg(抽象接口)
  --FixedTimeWinRateLimitAlg(实现类，就是重构前的RateLimitAlg)

RateLimiter的重构集中在构造函数

public class RateLimiter {
  private static final Logger log = LoggerFactory.getLogger(RateLimiter.class);
  // 为每个api在内存中存储限流计数器
  private ConcurrentHashMap<String, RateLimitAlg> counters = new ConcurrentHashMap<>();
  private RateLimitRule rule;

  public RateLimiter() {
    //改动主要在这里：调用RuleConfigSource类来实现配置加载
    RuleConfigSource configSource = new FileRuleConfigSource();
    RuleConfig ruleConfig = configSource.load();
    this.rule = new TrieRateLimitRule(ruleConfig);
  }

  public boolean limit(String appId, String url) throws InternalErrorException, InvalidUrlException {
    //...代码不变...
  }
}

修改后的代码（读取和解析的逻辑拆出来），有点策略模式的影子（根据文件格式选取不同的Parser）：

com.xzg.ratelimiter.rule.parser
  --RuleConfigParser(抽象接口)
  --YamlRuleConfigParser(Yaml格式配置文件解析类)
  --JsonRuleConfigParser(Json格式配置文件解析类)
com.xzg.ratelimiter.rule.datasource
  --RuleConfigSource(抽象接口)
  --FileRuleConfigSource(基于本地文件的配置类)
  
public interface RuleConfigParser {
  RuleConfig parse(String configText);
  RuleConfig parse(InputStream in);
}

public interface RuleConfigSource {
  RuleConfig load();
}

public class FileRuleConfigSource implements RuleConfigSource {
  private static final Logger log = LoggerFactory.getLogger(FileRuleConfigSource.class);

  public static final String API_LIMIT_CONFIG_NAME = "ratelimiter-rule";
  public static final String YAML_EXTENSION = "yaml";
  public static final String YML_EXTENSION = "yml";
  public static final String JSON_EXTENSION = "json";

  private static final String[] SUPPORT_EXTENSIONS =
      new String[] {YAML_EXTENSION, YML_EXTENSION, JSON_EXTENSION};
  private static final Map<String, RuleConfigParser> PARSER_MAP = new HashMap<>();

  static {
    PARSER_MAP.put(YAML_EXTENSION, new YamlRuleConfigParser());
    PARSER_MAP.put(YML_EXTENSION, new YamlRuleConfigParser());
    PARSER_MAP.put(JSON_EXTENSION, new JsonRuleConfigParser());
  }

  @Override
  public RuleConfig load() {
    for (String extension : SUPPORT_EXTENSIONS) {
      InputStream in = null;
      try {
        in = this.getClass().getResourceAsStream("/" + getFileNameByExt(extension));
        if (in != null) {
          RuleConfigParser parser = PARSER_MAP.get(extension);
          return parser.parse(in);
        }
      } finally {
        if (in != null) {
          try {
            in.close();
          } catch (IOException e) {
            log.error("close file error:{}", e);
          }
        }
      }
    }
    return null;
  }

  private String getFileNameByExt(String extension) {
    return API_LIMIT_CONFIG_NAME + "." + extension;
  }
}