dubbo 是如何进行限流的？

服务限流，为了防止服务的QPS突然飙升或者太大导致服务频繁失败甚至崩溃重启，对一些重要服务进行限流，虽然会导致一些请求失败，但是能够保证服务平稳运行 Dubbo中的限流有如下几个措施： accepts：限制服务器端接受的连接数，对于服务提供方，Dubbo默认基于Netty协议的底层网络可以限制客户端连接数量进行限制dubbo.protocol.accepts=10 或者dubbo.provider.accepts=10 只能在服务提供端 connections:限制客户端服务使用连接数,使用@DubboReference(connections = 10)或者 @DubboService(connections = 10) 作用在接口或方法级别上，如果两个都有，@DubboReference(connections = 10)优先 executes: 服务提供者并发限制，作用在服务提供者接口或者方法上,@DubboService(connections = 10,executes = 10)或@Method(executes = 10),Dubbo中会根据接口或者方法上的该配置，然后在每次执行方法前，会判断当前接口或者方法并行的数量是否大于该值，如果大于抛出异常，否则当前接口和方法active+1.在ExecuteLimitFilter实现了该过滤处理 actives:与executes类似，但是可以作用在服务提供者和消费者的接口或方法上，但是不同于executes不同的是，这里是表示同事能够活跃的处理数，如果超过该值，则需要等待timeoutms，超时则抛出异常，在ActiveLimitFilter实现了active处理

一般在分布式系统中，限流和熔断是处理并发的两大利器，一般来说客户端熔断，服务端限流。dubbo提供了多个和请求相关的filter：TPSLimiterFilter、ActiveLimitFilter、ExecuteLimitFilter，三者的侧重点各有不同 ### TpsLimitFilter dubbo中的限流通过TpsLimitFilter来实现，会在invoker执行实际业务逻辑前进行拦截，判断单位时间请求数是否超过上限，如果超过，抛出异常阻断调用。用于控制一段时间类中的请求数： ```java // PROVIDER服务端限制 @Activate(group = Constants.PROVIDER, value = Constants.TPS_LIMIT_RATE_KEY) public class TpsLimitFilter implements Filter { private final TPSLimiter tpsLimiter = new DefaultTPSLimiter(); @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { if (!tpsLimiter.isAllowable(invoker.getUrl(), invocation)) { throw new RpcException( "Failed to invoke service " + invoker.getInterface().getName() + "." + invocation.getMethodName() + " because exceed max service tps."); } return invoker.invoke(invocation); } } ``` 通过源码很明显看出，是通过TPSLimiter的isAllowable实现限流的。其内部采用了计数器算法，单位时间内限制多少调用次数，超过限制，返回false： ```java public class DefaultTPSLimiter implements TPSLimiter { // TPSLimiter针对每个service都创建一个计数器StatItem，通过StatItem的isAllowable方法判断请求是否有效 private final ConcurrentMap<String, StatItem> stats = new ConcurrentHashMap<String, StatItem>(); public boolean isAllowable(URL url, Invocation invocation) { int rate = url.getParameter(Constants.TPS_LIMIT_RATE_KEY, -1); long interval = url.getParameter(Constants.TPS_LIMIT_INTERVAL_KEY, Constants.DEFAULT_TPS_LIMIT_INTERVAL); // servicekey并没有和方法绑定，只能限流接口 String serviceKey = url.getServiceKey(); if (rate > 0) { StatItem statItem = stats.get(serviceKey); if (statItem == null) { stats.putIfAbsent(serviceKey, new StatItem(serviceKey, rate, interval)); statItem = stats.get(serviceKey); } return statItem.isAllowable(); } else { StatItem statItem = stats.get(serviceKey); if (statItem != null) { stats.remove(serviceKey); } } return true; } } ``` 查看StatItem类的部分代码，逻辑很简单，针对每段时间允许rate次调用，只要计数器达不到上限就返回true，超过时间重置计数器： ```java class StatItem { // 接口名 private String name; // 计数周期开始 private long lastResetTime; // 计数间隔 默认60s private long interval; // 剩余计数请求数 private AtomicInteger token; // 总共允许请求数 private int rate; public boolean isAllowable() { long now = System.currentTimeMillis(); if (now > lastResetTime + interval) { token.set(rate); lastResetTime = now; } int value = token.get(); boolean flag = false; while (value > 0 && !flag) { // 乐观锁增加计数 flag = token.compareAndSet(value, value - 1); value = token.get(); } return flag; } } ``` TpsLimitFilter是一个扩展点自动激活配置，首先TpsLimitFilter只对Provider端有效，其次Provider的url需要包括tps=xxx这个配置才能生效。而且Dubbo框架并没有默认通过配置文件启动这个Filter，所以我们需要在classpath的META-INF/dubbo/目录下增加com.alibaba.dubbo.rpc.Filter文件： ```java tps=com.alibaba.dubbo.rpc.filter.TpsLimitFilter ``` 前面说过，需要在url上配置tps参数才可以生效，如同降级一样，通过override动态配置覆盖规则，可以在代码中完成，也可以在监控中心或治理中心页面完成，因为在RegistryProtocol使用export方法对服务进行本地暴露以及注册Provider Url到zk后，还做了另外一个操作，监听服务对应的/dubbo/interface/configurations目录，一旦configurations目录下节点发生变化，就会重新生成暴露的url，然后进行reexport： ```java // 得到override url，用于监听configurations目录 final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl); // 构造监听器，用于provider url被override时，重新发布exporter // 监听路径为：/dubbo/interface/configurations final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker); overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener); // 向registry订阅这个url路径 registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener); ``` 比如最简单的：override://127.0.0.1:20880/com.test.TestService?category=configurators&tps=5&tps.interval=5000 ### ActiveLimitFilter dubbo中使用ActiveLimitFilter控制客户端同样的方法可同时并发请求数量，即该方法的并发度： ```java // CONSUMER客户端限制 @Activate(group = Constants.CONSUMER, value = Constants.ACTIVES_KEY) public class ActiveLimitFilter implements Filter { public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); // 获取设置的acvites的值，默认为0 int max = invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0); // 获取当前方法目前并发请求数量 RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()); if (max > 0) { // 等待结束依赖于timeout，需要搭配一起使用 long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0); long start = System.currentTimeMillis(); long remain = timeout; int active = count.getActive(); // 超过了指定的active值之后该请求将等待前面的请求完成 if (active >= max) { synchronized (count) { while ((active = count.getActive()) >= max) { try { count.wait(remain); } catch (InterruptedException e) { } long elapsed = System.currentTimeMillis() - start; remain = timeout - elapsed; // 如果等待时间超时，则抛出异常 if (remain <= 0) { throw new RpcException("Waiting concurrent invoke timeout in client-side for service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", elapsed: " + elapsed + ", timeout: " + timeout + ". concurrent invokes: " + active + ". max concurrent invoke limit: " + max); } } } } } // 没有限流正常调用 try { long begin = System.currentTimeMillis(); RpcStatus.beginCount(url, methodName); try { Result result = invoker.invoke(invocation); RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true); return result; } catch (RuntimeException t) { RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false); throw t; } } finally { if (max > 0) { synchronized (count) { count.notify(); } } } } } ``` 消费端并发控制配置(接口/方法)：<dubbo:service interface="com.foo.BarService" actives="10" />表示限制com.foo.BarService的每个方法，每客户端并发执行(或占用连接的请求数)不能超过10个 ### ExecuteLimitFilter 控制在单台服务上的并行度，单个方法的并发处理个数，与客户端控制区别是超过数量不会等待，而是直接抛出异常： ```java // PROVIDER服务端限制 @Activate(group = Constants.PROVIDER, value = Constants.EXECUTES_KEY) public class ExecuteLimitFilter implements Filter { public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); Semaphore executesLimit = null; boolean acquireResult = false; //获取executes值，默认为0 int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0); if (max > 0) { RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName()); // if (count.getActive() >= max) { /** * http://manzhizhen.iteye.com/blog/2386408 * use semaphore for concurrency control (to limit thread number) */ // 计数信号量设为max executesLimit = count.getSemaphore(max); // 如果并发处理数量大于设置的值，抛出异常 if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())) { throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited."); } } long begin = System.currentTimeMillis(); boolean isSuccess = true; RpcStatus.beginCount(url, methodName); try { Result result = invoker.invoke(invocation); return result; } catch (Throwable t) { isSuccess = false; if (t instanceof RuntimeException) { throw (RuntimeException) t; } else { throw new RpcException("unexpected exception when ExecuteLimitFilter", t); } } finally { RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess); if(acquireResult) { executesLimit.release(); } } } } ``` 服务端并发控制配置(接口/方法)：<dubbo:service interface="com.foo.BarService" executes="10" />表示限制com.foo.BarService的每个方法，服务器端并发执行(或占用线程池线程数)不能超过10个 ### 连接控制 限制服务器端接受的连接不能超过10个： ``` <dubbo:provider protocol="dubbo" accepts="10" /> 或 <dubbo:protocol name="dubbo" accepts="10" /> ``` 限制客户端服务使用连接不能超过10个： ``` <dubbo:reference interface="com.foo.BarService" connections="10" /> 优先于 <dubbo:service interface="com.foo.BarService" connections="10" /> ```

### Dubbo限流 Dubbo的限流作用于提供方。可以在高并发的情况，保证系统的稳定性、安全性。避免让系统被流量压垮，导致整体服务不可用。 ### 实践 提供者添加类似配置 ``` <dubbo:service interface="com.huang.yuan.api.service.DemoService" ref="demoServiceImpl" version="1.0" delay="5000" filter="tps"> <dubbo:parameter key="tps" value="1"/> <dubbo:parameter key="tps.interval" value="1000"/> </dubbo:service> ``` 添加filter及dubbo paramter，表示每tps.interval的时间间隔内，能执行tps个请求。 消费方同时发出10个请求 ``` @Test public void testdada() throws Exception { for (int i = 0; i < 10; i++) { new Thread(()->{ demoService.test("huangyuan"); }).start(); } Thread.sleep(1100000); } ``` 提供方限制流量，多余的请求将抛出异常：  ### 源码 ### 令牌桶算法 Dubbo默认使用令牌桶算法实现限流。某段时间内，桶里面只能放进n个令牌，然后来一个请求就减少一个令牌，如果桶里面的令牌没有了，则不能继续执行请求。 限流通过com.alibaba.dubbo.rpc.filter.TpsLimitFilter实现。 首先从URL中获取配置的限制，限制由两个参数组成，表示在interval毫秒的时间内允许执行rate个调用。 默认周期是60秒，不限制速率。 ``` public boolean isAllowable() { // 获取现在的时间 long now = System.currentTimeMillis(); // 当经过了interval时间间隔 if (now > lastResetTime + interval) { // 重新设置token令牌的个数 token.set(rate); // 从现在开始，经过interval的时间 lastResetTime = now; } // 获取令牌的值 int value = token.get(); boolean flag = false; // 使用CAS实现乐观锁 while (value > 0 && !flag) { // 能够执行请求，则令牌减一 flag = token.compareAndSet(value, value - 1); value = token.get(); } return flag; } ```