源码分析Dubbo tps过滤器器实现原理

微信公众号：[中间件兴趣圈]
作者简介：《RocketMQ技术内幕》作者

本文将重点分析一下dubbo限流的另外一个方式，tps过滤器。
@Activate(group = Constants.PROVIDER, value = Constants.TPS_LIMIT_RATE_KEY)

• 过滤器作用
  服务调用tps过滤器
• 使用场景
  对Dubbo服务提供者实现限流(tps)。
• 阻断条件
  当服务调用者超过其TPS时，直接返回rpc exception。

接下来从源码的角度分析Tps过滤器的实现机制。

1public class TpsLimitFilter implements Filter {
2    private final TPSLimiter tpsLimiter = new DefaultTPSLimiter();
3    @Override
4    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
5        if (!tpsLimiter.isAllowable(invoker.getUrl(), invocation)) {
6            throw new RpcException(
7                    "Failed to invoke service "  +  invoker.getInterface().getName() +  "." + invocation.getMethodName() + " because exceed max service tps.");
8        }
9        return invoker.invoke(invocation);
10    }
11}

tps limit 生效的条件是，服务提供者的url中包含了tps=""这个属性，默认TPS统计时长为1分钟，表示如果在1分钟之内的调用次数超过配置的tps，则阻断本次RPC服务调用。

其TPS控制代码主要由DefaultTPSLimiter实现。

1public class DefaultTPSLimiter implements TPSLimiter {
2    private final ConcurrentMap<String, StatItem> stats = new ConcurrentHashMap<String, StatItem>();
3    @Override
4    public boolean isAllowable(URL url, Invocation invocation) {
5        int rate = url.getParameter(Constants.TPS_LIMIT_RATE_KEY, -1);    // @1
6        long interval = url.getParameter(Constants.TPS_LIMIT_INTERVAL_KEY,
7                Constants.DEFAULT_TPS_LIMIT_INTERVAL);
8        String serviceKey = url.getServiceKey();                                            // @2
9        if (rate > 0) {
10            StatItem statItem = stats.get(serviceKey);
11            if (statItem == null) {
12                stats.putIfAbsent(serviceKey,
13                        new StatItem(serviceKey, rate, interval));
14                statItem = stats.get(serviceKey);
15            }
16            return statItem.isAllowable();                                                     // @3
17        } else {
18            StatItem statItem = stats.get(serviceKey);
19            if (statItem != null) {
20                stats.remove(serviceKey);
21            }
22        }
23
24        return true;
25    }
26}

代码@1：获取服务提供者url中的参数tps、tps.interval属性。

代码@2：获取服务key，并创建或获取对应的StatItem。

代码@3：调用StatItem的isAllowable()方法来判断是否可用。

StatItem#isAllowable

1public boolean isAllowable() {
2        long now = System.currentTimeMillis();
3        if (now > lastResetTime + interval) {    // @1
4            token.set(rate);
5            lastResetTime = now;
6        }
7
8        int value = token.get();
9        boolean flag = false;
10        while (value > 0 && !flag) {     // @2
11            flag = token.compareAndSet(value, value - 1);
12            value = token.get();
13        }
14        return flag;
15    }

该类的核心思想：是漏桶算法。
代码@1：如果当前时间大于（上一次刷新时间+统计间隔），重新复位token为rate，表示重新生成一批token。

代码@2：每使用一次，消耗一个token，如果能成功消耗一个token则返回true，如果没有可消耗的token，则直接返回false。

Tps过滤器的实现原理其实比较简单，大家可以从这里体会到ConcurrentHashMap、漏桶算法的简易实现。

广告：作者的新书《RocketMQ技术内幕》已上市

《RocketMQ技术内幕》已出版上市，目前可在主流购物平台（京东、天猫等）购买，本书从源码角度深度分析了RocketMQ NameServer、消息发送、消息存储、消息消费、消息过滤、主从同步HA、事务消息；在实战篇重点介绍了RocketMQ运维管理界面与当前支持的39个运维命令；并在附录部分罗列了RocketMQ几乎所有的配置参数。本书得到了RocketMQ创始人、阿里巴巴Messaging开源技术负责人、Linux OpenMessaging 主席的高度认可并作序推荐。目前是国内第一本成体系剖析RocketMQ的书籍。
新书7折优惠！7折优惠！7折优惠！
更多文章请关注微信公众号：

推荐关注微信公众号：RocketMQ官方微信公众号