gitee：
https://gitee.com/qiaodaimadewangcai/flood-myth
一、目标和需求分析
为了框架能满足当代互联网的基本需求，和使用的便利，优先实现以下几点需求。
- 支持分布式
- 支持SpringBoot-start
- 支持失败回调
- 单机模式下支持微秒级响应，分布式模式下支持毫秒级响应
其余还有写其他值得讨论实现的内容，优先级不是最高暂且先放着，我们以后有空再实现
- 支持高可靠性
- 支持监控
- 支持动态调节
- 支持持久化
二、初步设计
采用中心化的方式支持分布式，框架分成2部分“客户端”与”服务端”，为了方便还是将两部分写在一个工程中，客户端与服务端通过http进行通讯
三、客户端的实现
1）限流器参数定义
参数定义都是简单的pojo，不做过多说明，具体看代码
（1）限流规则定义
主要定义令牌桶的参数、限流器的运行模式和行为模式
https://gitee.com/qiaodaimadewangcai/flood-myth/blob/master/src/main/java/com/gyx/floodmyth/entity/LimiterRuleWrapper.java
（2）客户端配置定义
是一个单例模式，负责记录服务器的信息，包含一个线程池，用于向服务器发送心跳，并且拉取服务器上的信息
https://gitee.com/qiaodaimadewangcai/flood-myth/blob/master/src/main/java/com/gyx/floodmyth/entity/LimiterConfigWrapper.java
2）限流器实现
每个限流器都对应一个限流规则、一个客户端配置。
限流器中最重要的是实现一个限流算法，目前比较流行的几种限流算法——滑窗、漏桶、令牌桶。这里采用令牌桶限流。
令牌桶的实现主要包括2个部分
整个限流器其实都是在令牌桶的实现上添加了一些功能
(1)接口定义
限流器主要考虑初始化方法、尝试访问限流的方法
public interface LimiterHandler {
/**
* 初始化
* @param rule 限流规则的包装器
*/
void init(LimiterRuleWrapper rule);
/**
* 尝试访问
* @param tokenNum 消耗的令牌数量
*/
boolean tryAccess(Integer tokenNum);
/**
* 获取限流规则标识
*/
String getId();
/**
* 获取限流规则
*/
LimiterRuleWrapper getRule();
}
（2）限流器的抽象实现
我们想看成员变量的部分，除了对限流规则、客户端配置的持有，包括一个令牌的计数器（bucket），一个令牌桶的填装器（scheduledFuture）。
public abstract class AbstractLimiterHandler implements LimiterHandler {
/**
* 令牌桶
* 初始容量为0
*/
protected final AtomicLong bucket = new AtomicLong(0);
/**
* 限流规则
*/
protected LimiterRuleWrapper rule;
/**
* 限流器集群配置
*/
protected LimiterConfigWrapper config;
/**
* 令牌装填器
*
* 用于给令牌桶补充令牌
*/
protected ScheduledFuture<?> scheduledFuture;
}
限流器初始化的时候，必须填入限流规则和客户端配置，并且会停止令牌桶的装填
在限流规则发生改变的时候，可以单独调用init方法，以便用新的规则替换旧的规则
public AbstractLimiterHandler(LimiterRuleWrapper rule, LimiterConfigWrapper config) {
this.config = config;
init(rule);
}
/**
* 初始化
* @param rule 限流规则的包装器
*/
@Override
public void init(LimiterRuleWrapper rule) {
this.rule = rule;
if (this.scheduledFuture != null) {
this.scheduledFuture.cancel(true);
}
}
尝试访问的方法会消耗令牌，当limit==0的时候，意味着不会填装令牌，所以直接返回false。
AccessStrategy是一个访问策略接口，这里使用了策略模式，提供2中访问策略
/**
* 尝试访问
*
* @param tokenNum 消耗的令牌数量
*/
@Override
public boolean tryAccess(Integer tokenNum) {
if (rule.isEnable()) {
//限流功能已关闭
return true;
}
if (rule.getLimit() == 0) {
return false;
}
return AccessStrategy.strategy.get(rule.getAccessModel()).tryAccess(bucket, rule,tokenNum);
}
getId方法、getRule只是简单的get方法，前一个返回rule的id，后一个返回rule。
（2）访问策略实现
访问策略一共有2种，代码结构上通过策略模式进行解耦，以满足开闭原则。
接口中包含一个静态变量，和一个方法。静态变量实际上是一个简单工厂，用来初始化和访问不同的策略。
public interface AccessStrategy {
/**
* 用于访问策略
*/
Map<AccessModel, AccessStrategy> strategy = new HashMap<AccessModel, AccessStrategy>(2) {{
put(AccessModel.FAIL_FAST, new FailFastAccess());
put(AccessModel.BLOCKING, new BlockingAccess());
}};
/**
* 尝试访问
*
* @param bucket 令牌桶
* @param rule 限流器规则
* @param tokenNum 消耗的令牌数量
*/
boolean tryAccess(AtomicLong bucket, LimiterRuleWrapper rule, Integer tokenNum);
}
快速失败访问策略，令牌不够立马失败，返回失败
public class FailFastAccess implements AccessStrategy {
@Override
public boolean tryAccess(AtomicLong bucket, LimiterRuleWrapper rule, Integer tokenNum) {
//CAS获取令牌,没有令牌立即失败
long l = bucket.longValue();
while (l >= tokenNum) {
if (bucket.compareAndSet(l, l - tokenNum)) {
return true;
}
l = bucket.longValue();
}
return false;
}
}
阻塞访问策略，令牌不够的时候，阻塞线程，直到令牌足够
public class BlockingAccess implements AccessStrategy {
@Override
public boolean tryAccess(AtomicLong bucket, LimiterRuleWrapper rule,Integer tokenNum) {
//CAS获取令牌,阻塞直到成功
long l = bucket.longValue();
while (!(l >= tokenNum && bucket.compareAndSet(l, l - tokenNum))) {
sleep(rule);
l = bucket.longValue();
}
return true;
}
/** sleep方法 **/
}
（3）单机限流器的实现
单机限流器继承抽象限流器，也是通过父类的构造器进行初始化，这里需要注意父类的构造器中会调用init初始化方法，但是实际执行的init并非父类中的init方法，而是子类重写的init方法。
public class LocalLimiterHandler extends AbstractLimiterHandler {
public LocalLimiterHandler(LimiterRuleWrapper rule, LimiterConfigWrapper config) {
super(rule, config);
}
}
重写父类中的init方法，在父类的基础上额外初始化定时器，将限流规则中的参数，填入到线程池中就行了，定时器会按照指定的周期，定时的装填令牌。
/** * 初始化 * @param rule 限流规则的包装器 */@Overridepublic void init(LimiterRuleWrapper rule) { super.init(rule); if (rule.getLimit() == 0) { return; } this.scheduledFuture = config.getScheduledThreadExecutor() .scheduleAtFixedRate(() -> { //当前的令牌数 + 每次填装的令牌数 < 最大令牌数 if (bucket.get() + rule.getLimit() < rule.getMaxLimit()) { bucket.set(rule.getLimit()); } //首次延迟时间、周期单位时间、时间单位 }, rule.getInitialDelay(), rule.getPeriod(), rule.getUnit());}
其他方法都在抽象类中已经实现了不需要修改。
（4）分布式限流器的实现
分布式限流器一样继承抽象限流器，但是无需重写init方法，分布式限流器的令牌填装是通过与服务器连接完成的，所以不需要初始化定时器。
public class CloudLimiterHandler extends AbstractLimiterHandler { public CloudLimiterHandler(LimiterRuleWrapper rule, LimiterConfigWrapper config) { super(rule, config); }}
重写尝试访问的方法，分布式消耗令牌的逻辑也是在客户端上实现的，和单机的逻辑没有区别，在尝试访问结束之后，会访问服务器获取令牌，填装令牌桶。
/** * 尝试访问 * * @param tokenNum 消耗的令牌数量 */@Overridepublic boolean tryAccess(Integer tokenNum) { boolean accessFlag = super.tryAccess(tokenNum); putCloudBucket(); return accessFlag;}
获取令牌的方法，看似繁琐，实际上只是用客户端配置中的定时器执行一个http请求，获取到令牌后填装到令牌桶中，其余的全是判断，中间用到一个经典的双重检查锁。
/** * 从集群令牌分发中心，获取令牌，填装到令牌桶中 */private void putCloudBucket() { //校验令牌数量是否需要获取 if (bucket.get() * rule.getBatch() > rule.getRemaining()) { return; } //获取定时器线程 config.getScheduledThreadExecutor().execute(() -> { //双重检查锁 第一层 if (bucket.get() * rule.getBatch() <= rule.getRemaining()) { //双重检查锁 上锁 synchronized (bucket) { //双重检查锁 第二层 if (bucket.get() * rule.getBatch() <= rule.getRemaining()) { //发送http获取令牌，然后填装到令牌桶中 String result = config.getAllotServer().connect(LimiterConfigWrapper.http_token, JSON.toJSONString(rule)); if (result != null) { bucket.getAndAdd(Long.parseLong(result)); } } } } });}
看完上面的代码肯定对AllotServer还不太清楚，AllotServer是对服务器资源访问和管理的类，接下来就一起看一下这个类。
3）分发服务器管理类实现
AllotServer用于记录服务器的地址，和提供相应的访问方法。一共4个成员变量
- serverList——用于记录令牌分发服务器地址，读多写少
- backupsList——地址的备份，当服务器访问失败的后，地址会被转移到这里
- lock——读写锁，防止读取的地址的时候，地址发送变化
- pos——用于轮询的计数器
public class AllotServer { private List<String> serverList = new CopyOnWriteArrayList<>(); private List<String> backupsList = new CopyOnWriteArrayList<>(); private ReentrantLock lock = new ReentrantLock(); private int pos = 0;}
设置令牌分发服务器，map的key是服务器的ip地址，value是服务器的权重，先清空原本的列表，然后模仿CopyOnWriteArrayList，通过复制避免并发问题。
添加权重的方式，也采用比较偷懒的方式，像list中添加重复元素，权重越高的元素，重复的次数越多
public void setServer(Map<String, Integer> ip) { // 清空List serverList.clear(); // 重建一个Map，避免服务器的上下线导致的并发问题 Map<String, Integer> serverMap = new HashMap<>(ip); // 取得Ip地址List for (String server : serverMap.keySet()) { int weight = serverMap.get(server); //添加权重 for (int i = 0; i < weight; i++) { serverList.add(server); } }}
获取服务器方法，获取服务器地址的时候需要上锁，防止冲突，当地址全部失效后，从之前失效的地址再次尝试连接，通过轮询的方式对服务器进行访问。
private String getServer() { String server; lock.lock(); try { if (serverList.size()==0){ serverList.addAll(backupsList); backupsList.clear(); } if (pos >= serverList.size()) { pos = 0; } server = serverList.get(pos); pos++; } finally { lock.unlock(); } return server;}
访问服务器的方法，非常简单向服务器发送一个http请求就行了
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) {
serverList.remove(server);
backupsList.add(server);
}
return null;
}
4）限流器注册实现
限流器注册过程包括2部分，限流器的构造工厂和注册器。
（1）限流器构造工厂
限流器的构造工厂很简单，通过简单工厂进行创建，然后向注册器注册
public class LimiterFactory {
public static LimiterHandler of(LimiterRuleWrapper rule) {
return of(rule, LimiterConfigWrapper.getInstance());
}
public static LimiterHandler of(LimiterRuleWrapper rule, LimiterConfigWrapper config) {
switch (rule.getLimiterModel()) {
case LOCAL:
//本地限流
LimiterHandler limiter = new LocalLimiterHandler(rule, config);
RegisterServer.registered(limiter);
return limiter;
case CLOUD:
//集群限流
limiter = new CloudLimiterHandler(rule, config);
rule.setName(rule.getName() == null ? String.valueOf(limiter.hashCode()) : rule.getName());
RegisterServer.registered(limiter, config);
return limiter;
default:
throw new RuntimeException("无法识别限流处理器运行模式");
}
}
}
（2）限流注册器
注册器主要用于缓存所有的限流器，并且提供相应的访问方法
注册器通过一个map缓存所有的限流器，key为id，value为限流器实例。ConcurrentHashMap保证线程安全
public class RegisterServer {
/**
* 限流处理器的容器
*/
private static Map<String, LimiterHandler> limiterContainer = new ConcurrentHashMap<>();
}
提供一个静态方法，可以很方便的访问所有的限流器
public static LimiterHandler get(String id) {
LimiterHandler limiterHandler = limiterContainer.get(id);
if (limiterHandler == null){
throw new RuntimeException("无法查询到处理");
}
return limiterHandler;
}
本地限流器的注册非常简单，放入map就行了
public static void registered(LimiterHandler limiter) {
if (limiterContainer.containsKey(limiter.getId())) {
throw new RuntimeException("不可以重复注册限流处理器，限流器id:" + limiter.getId());
}
limiterContainer.put(limiter.getId(), limiter);
}
分布式限流器注册除了需要将对象存入map，还需要想服务器发出请求，同步服务器上的限流规则，如果连接失败，会转成本地服务运行。
/**
* 分布式注册
*
* @param limiter 限流处理器
* @param config 限流器配置包装类
*/
public static void registered(LimiterHandler limiter, LimiterConfigWrapper config) {
//注册在本地
registered(limiter);
//从令牌中心拉取规则，更新本地限流规则
rulePull(limiter, config);
}
/**
* 从令牌中心拉取规则，更新本地限流规则
*
* @param limiter 限流处理器
* @param config 限流器配置包装类
*/
private static void rulePull(LimiterHandler limiter, LimiterConfigWrapper config) {
config.getScheduledThreadExecutor().scheduleWithFixedDelay(() -> {
//连接远程获取配置
String rules = config.getAllotServer().connect(LimiterConfigWrapper.http_heart, JSON.toJSONString(limiter.getRule()));
if (rules == null) {
//连接失败，转成本地模式运行
LimiterRuleWrapper rule = limiter.getRule();
rule.setLimiterModel(LimiterModel.LOCAL);
limiter.init(rule);
return;
}
LimiterRuleWrapper newestRule = JSON.parseObject(rules, LimiterRuleWrapper.class);
if (newestRule.getVersion() > limiter.getRule().getVersion()) {
//版本升级
if (newestRule.getLimiterModel().equals(LimiterModel.LOCAL)) {
//禁止改成本地模式
newestRule.setLimiterModel(LimiterModel.CLOUD);
}
//更新规则
limiterContainer.get(limiter.getId()).init(newestRule);
}
}, 0, 1, TimeUnit.SECONDS);
}
5）注解实现
通过注解对接口进行限流，被注解的方法就会访问限流器进行限流，如果限流失败会调用指定的回调方法
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@Documented
@Inherited
public @interface Limiter {
/**
* Limiter id
*/
String value() default "";
/**
* 令牌消耗数量
*/
int num() default 1;
/**
* 回调方法
*/
String fallback() default "";
}
注解的实现很简单，通过环绕切面，将限流器调用包裹在目标方法外，如果执行失败，就调用回调方法，这个的回调方法比较简单，所以回调方法必须和注解注释的方法在同一个类中，并且参数完全一致
@Aspect
public class LimiterAspect {
@Pointcut("@annotation(com.gyx.floodmyth.aspect.Limiter)")
public void pointcut() {
}
@Around("pointcut() && @annotation(limiter)")
public Object around(ProceedingJoinPoint pjp, Limiter limiter) throws Throwable {
LimiterHandler rateLimiter = RegisterServer.get(limiter.value());
if (rateLimiter.tryAccess(limiter.num())) {
return pjp.proceed();
}
//快速失败后的回调方法
return fallback(pjp, limiter);
}
/**
* 快速失败的回调方法
* @param pjp 切入点
* @param limiter 注解数据
*/
private Object fallback(ProceedingJoinPoint pjp, Limiter limiter) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
Signature sig = pjp.getSignature();
if (!(sig instanceof MethodSignature)) {
throw new IllegalArgumentException("此注解只能使用在方法上");
}
//回调方法必须和注解注释的方法在同一个类中，并且参数完全一致
MethodSignature msg = (MethodSignature) sig;
Object target = pjp.getTarget();
Method fallback = target.getClass().getMethod(limiter.fallback(), msg.getParameterTypes());
return fallback.invoke(target, pjp.getArgs());
}
}
四、服务器实现
未完
