Talk about TPSLimiter of dubbo

  dubbo

Order

This article mainly studies dubbo’s TPSLimiter

TPSLimiter

dubbo-2.7.2/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/tps/TPSLimiter.java

public interface TPSLimiter {

    /**
     * judge if the current invocation is allowed by TPS rule
     *
     * @param url        url
     * @param invocation invocation
     * @return true allow the current invocation, otherwise, return false
     */
    boolean isAllowable(URL url, Invocation invocation);

}
  • TPSLimiter defines isAllowable method

DefaultTPSLimiter

dubbo-2.7.2/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/tps/DefaultTPSLimiter.java

public class DefaultTPSLimiter implements TPSLimiter {

    private final ConcurrentMap<String, StatItem> stats = new ConcurrentHashMap<String, StatItem>();

    @Override
    public boolean isAllowable(URL url, Invocation invocation) {
        int rate = url.getParameter(TPS_LIMIT_RATE_KEY, -1);
        long interval = url.getParameter(TPS_LIMIT_INTERVAL_KEY, DEFAULT_TPS_LIMIT_INTERVAL);
        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);
            } else {
                //rate or interval has changed, rebuild
                if (statItem.getRate() != rate || statItem.getInterval() != interval) {
                    stats.put(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;
    }

}
  • DefaultTPSLimiter implements TPSLimiter, which uses ConcurrentHashMap to store StatItem, and its key is serviceKey; in URL; IsAllowable method reads tps parameters from URL, default is -1, if it is less than 0, it will be removed from ConcurrentHashMap, if it is greater than 0, it will create or get StatItem, calling isAllowable (Reset or decrement token and return results)

StatItem

dubbo-2.7.2/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/tps/StatItem.java

class StatItem {

    private String name;

    private long lastResetTime;

    private long interval;

    private LongAdder token;

    private int rate;

    StatItem(String name, int rate, long interval) {
        this.name = name;
        this.rate = rate;
        this.interval = interval;
        this.lastResetTime = System.currentTimeMillis();
        this.token = buildLongAdder(rate);
    }

    public boolean isAllowable() {
        long now = System.currentTimeMillis();
        if (now > lastResetTime + interval) {
            token = buildLongAdder(rate);
            lastResetTime = now;
        }

        if (token.sum() < 0) {
            return false;
        }
        token.decrement();
        return true;
    }

    public long getInterval() {
        return interval;
    }


    public int getRate() {
        return rate;
    }


    long getLastResetTime() {
        return lastResetTime;
    }

    long getToken() {
        return token.sum();
    }

    @Override
    public String toString() {
        return new StringBuilder(32).append("StatItem ")
                .append("[name=").append(name).append(", ")
                .append("rate = ").append(rate).append(", ")
                .append("interval = ").append(interval).append("]")
                .toString();
    }

    private LongAdder buildLongAdder(int rate) {
        LongAdder adder = new LongAdder();
        adder.add(rate);
        return adder;
    }

}
  • StatItem defines a token of type LongAdder, whose isAllowable method determines whether the token needs to be reset, and if so, uses buildLongAdder to reset the token, or if not, intoken.sum() < 0Returns false if it is greater than or equal to 0, and decrements token if it is greater than or equal to 0.

Summary

  • TPSLimiter defines isAllowable method
  • DefaultTPSLimiter implements TPSLimiter, which uses ConcurrentHashMap to store StatItem, and its key is serviceKey; in URL; IsAllowable method reads tps parameters from URL, default is -1, if it is less than 0, it will be removed from ConcurrentHashMap, if it is greater than 0, it will create or get StatItem, calling isAllowable (Reset or decrement token and return results)
  • StatItem defines a token of type LongAdder, whose isAllowable method determines whether the token needs to be reset, and if so, uses buildLongAdder to reset the token, or if not, intoken.sum() < 0Returns false if it is greater than or equal to 0, and decrements token if it is greater than or equal to 0.

doc