5 posts tagged with "spi"

Gateway applications need to support a variety of load balancing strategies, including random,Hashing, RoundRobin and so on. In Apache Shenyu gateway, it not only realizes such traditional algorithms, but also makes smoother traffic processing for the entry of server nodes through detailed processing such as traffic warm-up, so as to obtain better overall stability. In this article, let's walk through how Apache Shenyu is designed and implemented this part of the function.

This article based on shenyu-2.5.0 version of the source code analysis.

[TOC]

LoadBalancer SPI#

The implementation of LoadBalancer is in shenyu-loadbalancer module. It has based on its SPI creation mechanism. The core interface code is shown as follows. This interface well explains the concept: load balancing is to select the most appropriate node from a series of server nodes. Routing, traffic processing and load balancing is the basic function of LoadBalancer SPI.

@SPIpublic interface LoadBalancer {
    /**     * this is select one for upstream list.     *     * @param upstreamList upstream list     * @param ip ip     * @return upstream     */    Upstream select(List<Upstream> upstreamList, String ip);}

Where upstreamList represents the server nodes list available for routing. Upstream is the data structure of server node, the important elements including protocol, upstreamUrl , weight, timestamp, warmup、healthy.

public class Upstream {    /**     * protocol.     */    private final String protocol;
    /**     * url.     */    private String url;
    /**     * weight.     */    private final int weight;
    /**     * false close, true open.     */    private boolean status;
    /**     * startup time.     */    private final long timestamp;
    /**     * warmup.     */    private final int warmup;
    /**     * healthy.     */    private boolean healthy;
    /**     * lastHealthTimestamp.     */    private long lastHealthTimestamp;
    /**     * lastUnhealthyTimestamp.     */    private long lastUnhealthyTimestamp;
    /**     * group.     */    private String group;
    /**     * version.     */    private String version;}

Design of LoadBalancer module#

The class diagram of LoadBalancer moduleisshown as follows.

We can draw the outline of LoadBalancer module from the class diagram:

1. The abstract class AbstractLoadBalancer implements the SPI LoadBalancer interface，and supplies the template methods for selection related, such as select(), selector()，and gives the calculation of weight.

2. Three implementation classes which inherit AbstractLoadBalancer to realize their own logic:

   • RandomLoadBalancer - Weight Random
   • HashLoadBalancer - Consistent Hashing
   • RoundRobinLoadBalancer -Weight Round Robin per-packet

3. The factory class LoadBalancerFactory provides public static method to be called.

   The implementation classes and algorithms are configurable. According to its specification, by adding profile in SHENYU_DIERECTORY directory, the data in profile should be key=value-class format, where the value-class will be load by the Apache Shenyu SPI class loader, and key value should be an name defined in LoadBalanceEnum.

random=org.apache.shenyu.loadbalancer.spi.RandomLoadBalancerroundRobin=org.apache.shenyu.loadbalancer.spi.RoundRobinLoadBalancerhash=org.apache.shenyu.loadbalancer.spi.HashLoadBalancer

The code of LoadBalanceEnum is as follows：

public enum LoadBalanceEnum {    /**     * Hash load balance enum.     */    HASH(1, "hash", true),
    /**     * Random load balance enum.     */    RANDOM(2, "random", true),
    /**     * Round robin load balance enum.     */    ROUND_ROBIN(3, "roundRobin", true);
    private final int code;    private final String name;    private final boolean support;}

AbstractLoadBalancer#

This abstract class implements the LoadBalancer interface and define the abstract method doSelect() to be processed by the implementation classes. In the template method select(), It will do validation first then call the doSelect() method.

public abstract class AbstractLoadBalancer implements LoadBalancer {    /**     * Do select divide upstream.     *     * @param upstreamList the upstream list     * @param ip           the ip     * @return the divide upstream     */    protected abstract Upstream doSelect(List<Upstream> upstreamList, String ip);
    @Override    public Upstream select(final List<Upstream> upstreamList, final String ip) {        if (CollectionUtils.isEmpty(upstreamList)) {            return null;        }        if (upstreamList.size() == 1) {            return upstreamList.get(0);        }        return doSelect(upstreamList, ip);    }}

When the timestamp of server node is not null, and the interval between current time and timestamp is within the traffic warm-up time, the formula for weight calculation is. $$ {1-1} ww = min(1,uptime/(warmup/weight)) $$ It can be seen from the formula that the final weight(ww) is proportional to the original-weight value. The closer the time interval is to the warmup time, the greater the final ww. That is, the longer the waiting time of the request, the higher the final weight. When there is no timestamp or other conditions, the ww is equal to the weight value of Upstream object.

The central of thinking about warm-upis to avoid bad performance when adding new server and the new JVMs starting up.

Let's see how the load balancing with Random, Hashing and RoundRobin strategy is implemented.

RandomLoadBalancer#

The RandomLoadBalancer can handle two situations:

1. Each node without weight, or every node has the same weight, randomly choose one.
2. Server Nodes with different weight, choose one randomly by weight.

Following is the random() method of RandomLoadBalancer. When traversing server node list, if the randomly generated value is less than the weight of node, then the current node will be chosen. If after one round traversing, there is no server node match, then it will choose one randomly. The getWeight(final Upstream upstream) is defined in AbstractLoadBalancer class.

    @Override    public Upstream doSelect(final List<Upstream> upstreamList, final String ip) {        int length = upstreamList.size();        // every upstream has the same weight?        boolean sameWeight = true;        // the weight of every upstream        int[] weights = new int[length];        int firstUpstreamWeight = getWeight(upstreamList.get(0));        weights[0] = firstUpstreamWeight;        // init the totalWeight        int totalWeight = firstUpstreamWeight;        int halfLengthTotalWeight = 0;        for (int i = 1; i < length; i++) {            int currentUpstreamWeight = getWeight(upstreamList.get(i));            if (i <= (length + 1) / 2) {                halfLengthTotalWeight = totalWeight;            }            weights[i] = currentUpstreamWeight;            totalWeight += currentUpstreamWeight;            if (sameWeight && currentUpstreamWeight != firstUpstreamWeight) {                // Calculate whether the weight of ownership is the same.                sameWeight = false;            }        }        if (totalWeight > 0 && !sameWeight) {            return random(totalWeight, halfLengthTotalWeight, weights, upstreamList);        }        return random(upstreamList);    }
    private Upstream random(final int totalWeight, final int halfLengthTotalWeight, final int[] weights, final List<Upstream> upstreamList) {        // If the weights are not the same and the weights are greater than 0, then random by the total number of weights.        int offset = RANDOM.nextInt(totalWeight);        int index = 0;        int end = weights.length;        if (offset >= halfLengthTotalWeight) {            index = (weights.length + 1) / 2;            offset -= halfLengthTotalWeight;        } else {            end = (weights.length + 1) / 2;        }        // Determine which segment the random value falls on        for (; index < end; index++) {            offset -= weights[index];            if (offset < 0) {                return upstreamList.get(index);            }        }        return random(upstreamList);    }

HashLoadBalancer#

In HashLoadBalancer, it takes the advantages of consistent hashing , that maps both the input traffic and the servers to a unit circle, or name as hash ring. For the requestedip address, with its hash value to find the node closest in clockwise order as the node to be routed. Let's see how consistent hashing is implemented in HashLoadBalancer.

As to the hash algorithms, HashLoadBalancer uses MD5 hash, which has the advantage of mixing the input in an unpredictable but deterministic way. The output is a 32-bit integer. the code is shown as follows:

private static long hash(final String key) {    // md5 byte    MessageDigest md5;    try {        md5 = MessageDigest.getInstance("MD5");    } catch (NoSuchAlgorithmException e) {        throw new ShenyuException("MD5 not supported", e);    }    md5.reset();    byte[] keyBytes;    keyBytes = key.getBytes(StandardCharsets.UTF_8);    md5.update(keyBytes);    byte[] digest = md5.digest();    // hash code, Truncate to 32-bits    long hashCode = (long) (digest[3] & 0xFF) << 24            | ((long) (digest[2] & 0xFF) << 16)            | ((long) (digest[1] & 0xFF) << 8)            | (digest[0] & 0xFF);    return hashCode & 0xffffffffL;}

Importantly, how to generate the hash ring and avoid skewness? Let's thedoSelect() method inHashLoadBalancer as follows:

    private static final int VIRTUAL_NODE_NUM = 5;
    @Override    public Upstream doSelect(final List<Upstream> upstreamList, final String ip) {        final ConcurrentSkipListMap<Long, Upstream> treeMap = new ConcurrentSkipListMap<>();        upstreamList.forEach(upstream -> IntStream.range(0, VIRTUAL_NODE_NUM).forEach(i -> {            long addressHash = hash("SHENYU-" + upstream.getUrl() + "-HASH-" + i);            treeMap.put(addressHash, upstream);        }));        long hash = hash(ip);        SortedMap<Long, Upstream> lastRing = treeMap.tailMap(hash);        if (!lastRing.isEmpty()) {            return lastRing.get(lastRing.firstKey());        }        return treeMap.firstEntry().getValue();    }

In this method, duplicated labels are used which are called "virtual nodes" (i.e. 5 virtual nodes point to a single "real" server). It will make the distribution in hash ring more evenly, and reduce the occurrence of data skewness.

In order to rescue the data sorted in the hash ring, and can be accessed quickly, we use ConcurrentSkipListMap of Java to store the server node lists ( with virtual nodes) and its hash value as key. This class a member of Java Collections Framework, providing expected average log(n) time cost for retrieve and access operations safely execute concurrent by multiple threads.

Furthermore, the method tailMap(K fromKey) of ConcurrentSkipListMap can return a view of portion of the map whose keys are greater or equal to the fromKey, and not need to navigate the whole map.

In the above code section, after the hash ring is generated, it uses tailMap(K fromKey) of ConcurrentSkipListMap to find the subset that the elements greater, or equal to the hash value of the requested ip, its first element is just the node to be routed. With the suitable data structure, the code looks particularly clear and concise.

Consistent hashing resolved the poor scalability of the traditional hashing by modular operation.

RoundRobinLoadBalancer#

The original Round-robin selection is to select server nodes one by one from the candidate list. Whenever some nodes has crash ( ex, cannot be connected after 1 minute), it will be removed from the candidate list, and do not attend the next round, until the server node is recovered and it will be add to the candidate list again. In RoundRobinLoadBalancer,the weight Round Robin per-packet schema is implemented.

In order to work in concurrent system, it provides an inner static class WeigthRoundRobin to store and calculate the rolling selections of each server node. Following is the main section of this class( removed remark )

protected static class WeightedRoundRobin {
    private int weight;
    private final AtomicLong current = new AtomicLong(0);
    private long lastUpdate;
    void setWeight(final int weight) {        this.weight = weight;        current.set(0);    }    long increaseCurrent() {        return current.addAndGet(weight);    }
    void sel(final int total) {        current.addAndGet(-1 * total);    }    void setLastUpdate(final long lastUpdate) {        this.lastUpdate = lastUpdate;    }}

Please focus on the these method:

• setWeight(final int weight) : set the current value by weight

• increaseCurrent(): Increment the current value by weight, and current set to 0.

• sel(final int total): decrement the current value by total

  Let's see how the weight factor being used in this round-robin selection?

  First it defines a two-level ConcurrentMap type variable named as methodWeightMap , to cache the server node lists and the rolling selection data about each server node.

private final ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>(16);

In this map, the key of first level is set to upstreamUrl of first element in server node list. The type of second object is ConcurrentMap<String, WeightedRoundRobin>, the key of this inner Map is the value upstreamUrlvariable of each server node in this server list, the value object is WeightedRoundRobin, used to trace the rolling selection data about each server node. As to the implementation class for the Map object, we use ConcurrentHashMap of JUC, a hash table supporting full concurrency of retrievals and high expected concurrency for updates.

In the second level of the map, the embedded static class - WeighedRoundRobin of each node is thread-safe, implementing the weighted RoundRobin per bucket. The following is the code of the doselect() method of this class.

@Overridepublic Upstream doSelect(final List<Upstream> upstreamList, final String ip) {    String key = upstreamList.get(0).getUrl();    ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);    if (Objects.isNull(map)) {        methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<>(16));        map = methodWeightMap.get(key);    }    int totalWeight = 0;    long maxCurrent = Long.MIN_VALUE;    long now = System.currentTimeMillis();    Upstream selectedInvoker = null;    WeightedRoundRobin selectedWeightedRoundRobin = null;    for (Upstream upstream : upstreamList) {        String rKey = upstream.getUrl();        WeightedRoundRobin weightedRoundRobin = map.get(rKey);        int weight = getWeight(upstream);        if (Objects.isNull(weightedRoundRobin)) {            weightedRoundRobin = new WeightedRoundRobin();            weightedRoundRobin.setWeight(weight);            map.putIfAbsent(rKey, weightedRoundRobin);        }        if (weight != weightedRoundRobin.getWeight()) {            // weight changed.            weightedRoundRobin.setWeight(weight);        }        long cur = weightedRoundRobin.increaseCurrent();        weightedRoundRobin.setLastUpdate(now);        if (cur > maxCurrent) {            maxCurrent = cur;            selectedInvoker = upstream;            selectedWeightedRoundRobin = weightedRoundRobin;        }        totalWeight += weight;    }    ......  //erase the section which handles the time-out upstreams.     if (selectedInvoker != null) {        selectedWeightedRoundRobin.sel(totalWeight);        return selectedInvoker;    }    // should not happen here    return upstreamList.get(0);}

For example we assume upstreamUrl values of three server nodes is: LIST = [upstream-20, upstream-50, upstream-30]. After a round of execution, the data in newly created methodWeightMap is as follows:

For the above example LIST, assumes the weight array is [20,50,30]. the following figure shows the value change and polling selection process of the current array in WeighedRoundRobin object.

In each round, it will choose the server node with max current value.

• Round1:
  • Traverse the server node list, initialize the weightedRoundRobin instance of each server node or update the weight value of server nodes object Upstream
  • Traverse the server node list, initialize the weightedRoundRobin instance of each server node or update the weight value of server nodes object Upstream
  • say, in this case, after traverse, the current array of the node list changes to [20, 50,30]，so according to rule, the node Stream-50 would be chosen, and then the static object WeightedRoundRobin of Stream-50 executes sel(-total) , the current array is now [20,-50, 30].
• Round 2: after traverse, the current array should be [40,0,60], so the Stream-30 node would be chosen， current array is now [40,0,-40].
• Round 3: after traverse, current array changes to [60,50,-10], Stream-20 would be chosen，and current array is now [-40,50,-10].

When there is any inconsistence or some server crashed, for example, the lists size does not match with the elements in map, it would copy and modify the element with lock mechanism, and remove the timeout server node, the data in Map updated. Following is the fault tolerance code segment.

    if (!updateLock.get() && upstreamList.size() != map.size() && updateLock.compareAndSet(false, true)) {        try {            // copy -> modify -> update reference.            ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<>(map);            newMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > recyclePeriod);            methodWeightMap.put(key, newMap);        } finally {            updateLock.set(false);        }    }    if (Objects.nonNull(selectedInvoker)) {        selectedWeightedRoundRobin.sel(totalWeight);        return selectedInvoker;    }    // should not happen here.    return upstreamList.get(0);

LoadBalancerFactory#

In this class, a static method calling LoadBalancer is provided, whereExtensionLoader is the entry point of Apache Shenyu SPI. That is to say, LoadBalancer module is configurable and extensible. The algorithm variable in this static method is the name enumeration type defined in LoadBalanceEnum.

    /**     * Selector upstream.     *     * @param upstreamList the upstream list     * @param algorithm    the loadBalance algorithm     * @param ip           the ip     * @return the upstream     */    public static Upstream selector(final List<Upstream> upstreamList, final String algorithm, final String ip) {        LoadBalancer loadBalance = ExtensionLoader.getExtensionLoader(LoadBalancer.class).getJoin(algorithm);        return loadBalance.select(upstreamList, ip);    }

Using of LoadBalancer module#

In the above section, we describe the LoadBalancer SPI and three implementation classes. Let's take a look at how the LoadBalancer to be used in Apache Shenyu. DividePlugin is a plugin in Apache Shenyu responsible for routing http request. when enable to use this plugin, it will transfer traffic according to selection data and rule data, and deliver to next plugin downstream.

@Overrideprotected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {   ......}

The type of second parameter of doExecute() is ShenyuPluginChain, which represents the execution chain of plugins. For details, see the mechanism of Apache Shenyu Plugins. The third one is SelectorData type, and the fourth is RuleData type working as the rule data.

In doExecute() of DividePlugin, first verify the size of header, content length, etc, then preparing for load balancing.

Following is a code fragment usingLoadBalancer in the doExecute() method:

    // find the routing server node list    List<Upstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());    ...     // the requested ip    String ip = Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();
    //calling the Utility class and invoke the LoadBalance processing.    Upstream upstream = LoadBalancerFactory.selector(upstreamList, ruleHandle.getLoadBalance(), ip);

In the above code, the output ofruleHandle.getLoadBalance() is the name variable defined in LoadBalanceEnum, that is random, hash, roundRobin, etc. It is very convenient to use LoadBalancer by LoadBalancerFactory. When adding more LoadBalancer implementing classes, the interface in plugin module will not be effect at all.

Summary#

After reading through the code of LoadBalancer module, from the design perspective, it is concluded that this module has the following characteristics:

1. Extensibility: Interface oriented design and implemented on Apache Shenyu SPI mechanism, it can be easily extended to other dynamic load balancing algorithms (for example, least connection, fastest mode, etc), and supports cluster processing.
2. Scalability： Every load balancing implementation, weighted Random, consistency Hashing and weighted RoundRobin can well support increase or decrease cluster overall capacity.
3. More detailed design such as warm-up can bring better performance and obtain better overall stability.

In most of the plugins ( such as Dubbo, gRPC,Spring-cloud, etc) of Apache Shenyu, the routingparameters are designed to support the combination of multiple conditions. In order to realize such requirements, the parameters and behaviors are abstracted to three parts according to its SPI mechanism, and implemented in shenyu-plugin-base module.

• ParameterData-parameters
• PredictJudge-predicate
• MatchStrategy-matching strategy

Relatively speaking, the MatchStrategy is the part that needs the least extension points. For the combined judgement of multiple conditions, the common selection rules are: All conditions are matched, at least one is matched, at least the first is met, or most of conditions satisfied. As we will need to handle various types of parameters, for example: IP, header, uri, etc.

How to make the MatchStrategy to be simple to use and extensible?

MatchStrategy#

The implementation of MatchStrategy is in shenyu-plugin-base module. It is based on the SPI creation mechanism, and has used factory pattern and strategy design pattern. The class diagram of MatchStrategy is showed as follows.

Based on the interface MatchStrategy we design the implementation classes, and the abstract class AbstractMatchStrategy supplies common method, while the factory class MatchStrategyFactory provides creation functions.

MatchStrategy Interface#

First, let's look at the MatchStrategy SPI interface

@SPIpublic interface MatchStrategy {
    Boolean match(List<ConditionData> conditionDataList, ServerWebExchange exchange);}

The annotation @SPI means that this is an SPI interface. Where ServerWebExchange is org.springframework.web.server.ServerWebExchange, represents the request-response interactive content of HTTP. Following is the code of ConditionData, the more detail about this class can refer to code analysis of PredicteJudge

public class ConditionData {
    private String paramType;    private String operator;
    private String paramName;    private String paramValue;}

AbstractMatchStrategy#

Second, let's look at the abstract class AbstractMatchStrategy，it has defined a buildRealData method，In this method it wraps various parameters to a unified interface through the functionality of ParameterDataFactory, which is the factory class of ParameterData. It supports a variety of types of parameters , such as Ip, Cookie, Header,uri, etc. Modifications of such parameters will not impact the calling of matching rules of MatchStrategy.

public abstract class AbstractMatchStrategy {
    public String buildRealData(final ConditionData condition, final ServerWebExchange exchange) {        return ParameterDataFactory.builderData(condition.getParamType(), condition.getParamName(), exchange);    }}

Implementation class and profile#

Now, let's look at the two implementation class based on the above interface in shenyu-plugin-base module , that is:

• AndMatchStrategy- AND -All conditions are matched

• OrMatchStrategy- OR -at least one is match

  The properties file containing the SPI implementation is shown as follows, which located at the SHENYU_DIRECTORYdirectory. When starting up, the top-level SPI classes will read the key-value and load the classes and cache them.

and=org.apache.shenyu.plugin.base.condition.strategy.AndMatchStrategyor=org.apache.shenyu.plugin.base.condition.strategy.OrMatchStrategy

These two implementation classes inherit AbstractMatchStrategy class and implement MatchStrategy interface.

AndMatchStrategy- “AND” relation#

Since the PredicateJudge interface can encapsulate different variety of Predicates , for example EqualsPredicateJudge, EndsWithPredicateJudge and so on, the ConditionData and ParamData passed to it can present with variety of parameters, for treating of multiple conditions. So usingstream and lambda expression, it can be very simple and efficient to process "AND" logic (all conditions must be matched).

@Joinpublic class AndMatchStrategy extends AbstractMatchStrategy implements MatchStrategy {
    @Override    public Boolean match(final List<ConditionData> conditionDataList, final ServerWebExchange exchange) {        return conditionDataList                .stream()                .allMatch(condition -> PredicateJudgeFactory.judge(condition, buildRealData(condition, exchange)));    }}

The OrMatchStrategy similarly implements the "OR" logic- at least one is match.

MatchStrategyFactory#

This is the factory class of MatchStrategy，there are two methods, one is newInstance(), which will return the MatchStrategy implementation class instance cached by the SPI ExtensionLoader indexed by the key-value.

    public static MatchStrategy newInstance(final Integer strategy) {        String matchMode = MatchModeEnum.getMatchModeByCode(strategy);        return ExtensionLoader.getExtensionLoader(MatchStrategy.class).getJoin(matchMode);    }

the matchMode will be the name of strategy, the value will be "and" or "or". The MatchModeEnum defines the code and name of match strategy as follows.

AND(0, "and"), OR(1, "or");

Another method is match() method, which will invoke the match() method of implementation class.

    public static boolean match(final Integer strategy, final List<ConditionData> conditionDataList, final ServerWebExchange exchange) {        return newInstance(strategy).match(conditionDataList, exchange);    }

How it works#

AbstractShenyuPlugin is the base class of plugins in shenyu-plugin module. In this class two selection method are defined: filterSelector() and filterRule() , Both of them call the match() method of MatchStrategyFactory. The code of filterSelector() is shown as follows.

    private Boolean filterSelector(final SelectorData selector, final ServerWebExchange exchange) {        if (selector.getType() == SelectorTypeEnum.CUSTOM_FLOW.getCode()) {            if (CollectionUtils.isEmpty(selector.getConditionList())) {                return false;            }            return MatchStrategyFactory.match(selector.getMatchMode(), selector.getConditionList(), exchange);        }        return true;    }

In filterSelector() method, after validation of the SelectorData, calls the match method of MatchStrategyFactory, and then this factory class will invokes the match method of corresponding implementation class.

    private Boolean filterRule(final RuleData ruleData, final ServerWebExchange exchange) {        return ruleData.getEnabled() && MatchStrategyFactory.match(ruleData.getMatchMode(), ruleData.getConditionDataList(), exchange);    }

In filterRule() it is also calls the match() method of MatchStrategyFactory. Does it look particularly concise or even simple? In the code analysis of PredicteJudge , you can see more detail about parameter processing in shenyu-plugin.

Summary#

Due to the use of SPI mechanism of Apache Shenyu, the parameter selection module has the characteristic of loose coupling and extensibility. In terms of the combination of multiple conditions, MatchStrategy provides a good design. Although currently only two implementation classes are present, it can be easily used to develop more complex MatchStrategy rules in the future, such as "firstOf"-first condition must matched, or "mostOf"- most of the conditions must be matched, etc.

Interested readers can read the source code of 'shenyu-plugin' to learn more.

Apache Shenyu has been identified as a gateway application which supports a variety of protocols and microservice frameworks such as Dubbo, gRPC, Spring-Cloud, etc. To do this, the product has accomplished an elegant SPI (Service Provider Interface) as its foundation, and make the Rule data parsing and predicting program very simple , resiliency and security. As to rule data parsing processing, the SPI design increases the product's scalability. When appending new plugin, in most cases, the existing module is enough for rule data parsing , otherwise it can be rapidly carry out with tiny effort.

Top level design of SPI#

In Apache Shenyu, the SPI archtecure is defined in shenyu-spi module and composed of three parts: SPI interface, factory design pattern, and configuration file. There is two interface defined as annotation: @SPI and @Join. When class file with @Join annotation, it means that it will join as an SPI extension class, in other words, it is an application or registration. The @SPI denotes that the class is an SPI extension class.

Fig 1 classes in the shenyu-spi

The SPI configuration directory is META-INF/shenyu/. that is specified:

SHENYU_DIRECTORY = "META-INF/shenyu/";

When starting the gateway system , the ExtensionLoader will scan the profiles under SHENYU_DIRECTORY, in turn, load and validate and then initialize each configed class. The configuration file uses "Key = class-file" format. During operation of the system, the corresponding SPI implementation class will be invoked through the factory mechanism.

Implementation of shenyu-plugin SPI#

In shenyu-plugin module, various plugins for HTTP routing are implemented according to the plugin mechanism, including request, redirect, response and rewrite, etc. Plugins for microservice frameworks such as Dubbo, gRPC , Spring-Cloud and Tars have been developed in the gateway product. And plugins are still increasing. If no such dependent module fo parsing and judge routing parameters and data, each plugin is necessary to implement the parsing functions, and has to frequently modify to support their matching rules, such as wildcard, regular expression, SpEL expression, etc. Therefore, they made a high level abstraction for routing parameter data following the SPI framework in shenyu-plugin module. The rule analysis consists of three parts:

• ParameterData- parameter data

• PredicatJudge- predicate whether the actural data match the rule

• MatchStrategy- combine multiple conditions, the final used strategy

These implementation classes are defined in shenyu-plugin-base module. In each plugin, resolution and predication of the routing parameter can be realized through AbstractShenyuPlugin using the above SPIs. That is dedicated and easy to extend, in line with SOLID principle.

​ This section analyzes the PredictJudge in detail. You can find the dependency to shenyu-spi in the pom.xml of this module.

<dependency>    <groupId>org.apache.shenyu</groupId>    <artifactId>shenyu-spi</artifactId>    <version>${project.version}</version></dependency>

Design of PredicateJudge SPI#

PredicateJudge SPI is used to analyze and judge various routing rules configed in Apache Shenyu gateway. The name and functions of this SPI are similar to Predicate in Java, but the acceptance behavior is further abstracted applying for routing aspect. This SPI is implemented through the Factory pattern. Let's look at the Predictejudge SPI interface:

@SPI@FunctionalInterfacepublic interface PredicateJudge {
    /**     * judge conditionData and realData is match.     *     * @param conditionData {@linkplain ConditionData}     * @param realData       realData     * @return true is pass  false is not pass.     */    Boolean judge(ConditionData conditionData, String realData);}

The class diagram is as follows:

Fig 2-Predicate class diagram

The important methods of PredicateJudgeFactory are shown as follows:

Whenever need to parsing and matching routing data, you can use

    public static PredicateJudge newInstance(final String operator) {        return ExtensionLoader.getExtensionLoader(PredicateJudge.class).getJoin(processSpecialOperator(operator));    }

    public static Boolean judge(final ConditionData conditionData, final String realData) {        if (Objects.isNull(conditionData) || StringUtils.isBlank(realData)) {            return false;        }        return newInstance(conditionData.getOperator()).judge(conditionData, realData);    }

ConditionData contains of four attributes of String type: paramType, operator,paramName,paramValue

ParamTypeEnum#

Where paramType must be the enumeration type ParamTypeEnum. The default supported paramType are:

post, uri,query, host, ip,header, cookie,req_method

OperatorEnum#

operator must be the enumeration type OperatorEnum, currently supported operators are:

   match, =,regex, >,<, contains, SpEL,  Groovy, TimeBefore,TimeAfter

Base on the above defination , the plugin module provides the following eight PredicateJudge implemetation classes to realize the logic of these operators respectively.

How to use PredicateJudge#

When you want to parse parameters, you only need to call PredicateJudgeFactory as follows.

PredicateJudgeFactory.judge(final ConditionData conditionData, final String realData);

SPI profile#

The implementation class is configed in the file under directory SHENYU_DIRECTORY . It will be loaded and cached at startup.

equals=org.apache.shenyu.plugin.base.condition.judge.EqualsPredicateJudge
contains=org.apache.shenyu.plugin.base.condition.judge.ContainsPredicateJudgeGroovy=org.apache.shenyu.plugin.base.condition.judge.GroovyPredicateJudgematch=org.apache.shenyu.plugin.base.condition.judge.MatchPredicateJudgeregex=org.apache.shenyu.plugin.base.condition.judge.RegexPredicateJudgeSpEL=org.apache.shenyu.plugin.base.condition.judge.SpELPredicateJudgeTimeAfter=org.apache.shenyu.plugin.base.condition.judge.TimerAfterPredicateJudgeTimeBefore=org.apache.shenyu.plugin.base.condition.judge.TimerBeforePredicateJudge

The usage of PredicateJudge SPI in Shenyu gateway#

Most plugins in Apache Shenyu are inherited from AbstractShenyuPlugin. In this abstract class, the filter functions (selection and matching) are achieved through MatchStrategy SPI, and PredicateJudge will be invoked from MatchStrategy to predicate each condition data.

Fig 3- class diagram of plugins with PredicateJudge and MatchStrategy SPI

The process from client request calling the routing parsing moodule is showed as following chart.

Fig 4- flow chart for Shenyu gateway filter with parameter processing

• When startup, the system will load SPI classes from profile and cache them.
• When the client sends a new request to the Shenyu gateway, will call the corresponding plugin within the gateway.
• When analyzing real data with routing rules, the PredicateJudge implementation class will be invoked according to the contained operator.

Others#

Examples of PredicateJudge judgement#

ContainsPredicateJudge- " contains“ rule#

For example, giving a ConditionData with: paramType="uri", paramValue 是 "/http/**", when using the "contains" relation: ContainsPredicateJudge, the matching result is as follows.

About other PredicateJudge implemetantion classes, you can refer to the code and test classes.

Rate limiter is a very important integral of gateway application, to deal with high traffic. When the system is attacked abnormally by a large number of traffic gathered in a short time; When there are a large number of lower priority request need to be slow down or else it will effect your high priority transactions; Or sometimes your system can not afford the regular traffic; in these scenarios, we need to start rate limiter component to protect our system, through rejection, wait, load shedding,etc, limit the requests to an acceptable quantities, or only certain domains (or services) requests can get through.

Facing above scenarios, following need to be considered when designing the rate limiter component of an gateway.

1. Supports a variety of rate limiter algorithms and easy to extends.
2. Resilient resolvers which can distinguish traffic by different way, such as ip, url, even user group etc.
3. High availability, can quickly get allow or reject result from rate limiter
4. With fault tolerance against when rate limiter is down, the gateway can continue work.

This article will first introduce the overall architecture of the rate limiter module in Apache Shenyu, and then focus on the code analysis of rate limiter SPI.

This article based on shenyu-2.4.0 version of the source code analysis.

Overall design of RateLimiter#

Spring WebFlux is reactive and non-blocking web framework, which can benefit throughput and make applications more resilient. The plugin of Apache Shenyu is based on WebFlux，its rate limiter component is implemented in ratelimiter-plugin. In rate limiter process, the commonly used algorithms are token bucket, leaky bucket, etc. To speed up concurrency performance, the counting and calculation logic is treated in Redis, and Java code is responsible for the transmission of parameters. When applying Redis, the Lua script can be resident memory, and be executed as a whole, so it is atomic. Let alone the reducing of network overhead. Redis commands abstraction and automatic serialization/deserialization with Redis store is provided in Spring Data Redis. Because of based on reactive framework, the Spring Redis Reactive is used in ratelimiter-plugin.

The class diagram of this plugin is as follows, highlighting two packages related to RateLimiter SPI: resolver 和algorithm.

Design of RateLimiter SPI#

High performance issue is achieved through the architecture of Spring data+ Redis+Lua , two SPI are supplied in ratelimiter-plugin for the extension of algorithm and key resolver。

• RateLimiterAlgorithm：used for algorithms expansion.
• RateLimiterKeyResolver： used for resolver expansion, to distinguish requests by various information, including ip, url, ect.

The profile of SPI is located at directory of SHENYU_DIRECTORY (default/META-INF/shenyu).

RateLimiterKeyResolver#

Obtain the critical info of the request used for packet rate limiter，the interface of RateLimiterKeyResolver is follows：

@SPIpublic interface RateLimiterKeyResolver {
    /**     * get Key resolver's name.     *     * @return Key resolver's name     */    String getKeyResolverName();
    /**     * resolve.     *     * @param exchange exchange the current server exchange {@linkplain ServerWebExchange}     * @return rate limiter key     */    String resolve(ServerWebExchange exchange);}

@SPI registers the current interface as Apache Shenyu SPI. Method resolve(ServerWebExchange exchange) is used to provide the resolution way. Currently there are two key resolvers in RateLimiterKeyResolver SPI:WholeKeyResolve and RemoteAddrKeyResolver. The resolve method of RemoteAddrKeyResolveris as follows：

    @Override    public String resolve(final ServerWebExchange exchange) {        return Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();    }

Where the resolved key is ip of request. Based on SPI mechanism and its factory pattern, new resolver can be easily developed.

RateLimiterAlgorithm SPI#

RateLimiterAlgorithm SPI is used to identify and define different rate limiter algorithms, following is the class diagram of this module.

In this module, factory pattern is used , providing interface, abstract class and factory class, and four implementation classes. The lua script corresponding to the implementation class is enumerated in RateLimitEnum and located in /META-INF/scripts.

@SPIpublic interface RateLimiterAlgorithm<T> {        RedisScript<T> getScript();    List<String> getKeys(String id);        /**     * Callback string.     *     * @param script the script     * @param keys the keys     * @param scriptArgs the script args     */    default void callback(final RedisScript<?> script, final List<String> keys, final List<String> scriptArgs) {    }}

@SPI registers the current interface as Apache Shenyu SPI. There are three methods:

• getScript() returns a RedisScript object, which will be passed to Redis.
• getKeys(String id) returns a List contains with keys.
• callback() the callback function which will be executed asynchronously later on, and default is an empty method.

AbstractRateLimiterAlgorithm#

The template method is implemented in this abstract class, and the reified generics used is List<Long>. Two abstract methods getScriptName() and getKeyName() are left for the implementation class. Following is the code to load lua script.

    public RedisScript<List<Long>> getScript() {        if (!this.initialized.get()) {            DefaultRedisScript redisScript = new DefaultRedisScript<>();            String scriptPath = "/META-INF/scripts/" + getScriptName();            redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(scriptPath)));            redisScript.setResultType(List.class);            this.script = redisScript;            initialized.compareAndSet(false, true);            return redisScript;        }        return script;    }

initialized is an AtomicBoolean type variable used to indicate whether the lua script is loaded. If has not been loaded, the system will read specified scripts form META-INF/scripts; After reading, specify the result with List type, and set initialized=true, then returning RedisScriptobject.

The code of getKeys() in AbstractRateLimiterAlgorithm is as follows:

    @Override    public List<String> getKeys(final String id) {        String prefix = getKeyName() + ".{" + id;        String tokenKey = prefix + "}.tokens";        String timestampKey = prefix + "}.timestamp";        return Arrays.asList(tokenKey, timestampKey);    }

Two strings are generated in this template method, where the tokenKey will work as Key to a Sorted map in Redis.

We can observe from above class diagram that ConcurrentRateLimiterAlgorithm and SlidingWindowRateLimiterAlgorithm override getKeys(String id) method but another two implementation classes not, and use template method in AbstractRateLimiterAlgorithm. Only in ConcurrentRateLimiterAlgorithm has override callback() method, the others not. We will do further analysis in the following.

RateLimiterAlgorithmFactory#

The method getsRateLimiterAlgorithm instance by name in RateLimiterAlgorithmFactory is as follows:

public static RateLimiterAlgorithm<?> newInstance(final String name) {    return Optional.ofNullable(ExtensionLoader.getExtensionLoader(RateLimiterAlgorithm.class).getJoin(name)).orElse(new TokenBucketRateLimiterAlgorithm());}

ExtensionLoader of SPI is responsible for loading SPI classes by "name", if cannot find the specified algorithm class, it will return TokenBucketRateLimiterAlgorithm by default.

Data access with Redis#

Above detailed the extension interface in RateLimiter SPI. In Apache Shenyu, we use ReactiveRedisTemplate to perform Redis processing reactively, which is implemented inisAllowed() method of RedisRateLimiter class.

    public Mono<RateLimiterResponse> isAllowed(final String id, final RateLimiterHandle limiterHandle) {        // get parameters that will pass to redis from RateLimiterHandle Object        double replenishRate = limiterHandle.getReplenishRate();        double burstCapacity = limiterHandle.getBurstCapacity();        double requestCount = limiterHandle.getRequestCount();        // get the current used RateLimiterAlgorithm        RateLimiterAlgorithm<?> rateLimiterAlgorithm = RateLimiterAlgorithmFactory.newInstance(limiterHandle.getAlgorithmName());                ........        Flux<List<Long>> resultFlux = Singleton.INST.get(ReactiveRedisTemplate.class).execute(script, keys, scriptArgs);        return resultFlux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L)))                .reduce(new ArrayList<Long>(), (longs, l) -> {                    longs.addAll(l);                    return longs;                }).map(results -> {                    boolean allowed = results.get(0) == 1L;                    Long tokensLeft = results.get(1);                    return new RateLimiterResponse(allowed, tokensLeft);                })                .doOnError(throwable -> log.error("Error occurred while judging if user is allowed by RedisRateLimiter:{}", throwable.getMessage()))                .doFinally(signalType -> rateLimiterAlgorithm.callback(script, keys, scriptArgs));    }

The POJO class RateLimiterHandle wraps the parameters needed in rate limiter, they are algorithName, replenishRate, burstCapacity, requestCount, etc. First, gets the parameters that need to be passed into Redis from RateLimiterHandle class. Then obtain the current implementation class from RateLimiterAlgorithmFactory.

For convenience, we give an flow image to show the parameters I/O and execution procedure in Java and Redis respectively. On the left is the second half of isAllowed() , and on the right is the processing of Lua script.

Following is the execution process of the JAVA code.

1. Get two keys value in List<String> type from the getKeys() method, the first element will map to a sorted set in Redis.

2. Set four parameters, replenishRate , burstCapacity, timestamp (EpochSecond) and requestcount.

3. Calling ReactiveRedisTemplate with the scripts, keys and parameters, the return a Flux<List<Long>>

4. The return value is converted from Flux<ArrayList<Long>> to Mono<ArrayList<Long>> the through reduce() of Flux ，and then transform it to Mono<RateLimiterResponse> via map() function. Returned two data, one is allowed (1-allow, 0- not allowed), the other is tokensLeft, the number of available remaining request.

5. As for the fault tolerance, due to using of reactor non-blocking model, when an error occurs, the fallback function onErrorResume() will be executed and a new stream (1L, -1L) will generated by Flux.just, which means allow the request getting through, and log the error on the side.

6. After that, performs the doFinally() method, that is to execute the callback() method of the implementation class.

Four rate limiter algorithms#

From above we know that how the java code works with Redis in the gateway. In this chapter we briefly analysis some code of the four rate limiter algorithms, to understand how to develop the interface of RateLimiter SPI and work efficiently with Redis.

Four rate limiter algorithms are supplied in Apache Shenyu Ratelimit SPI:

1. Token bucket rate limiter： Limiting the traffic according to the number of requests. Assuming that N requests can be passed per second, when requests exceeding N will be rejected. In implementing of the algorithm, the requests will be grouped by bucket, the tokens will be generated at an evenly rate. If the number of requests is less than the tokens in the bucket, then it is allowed to pass. The time window is 2* capacity/rate.
2. Slide window rate limiter: Different from token bucket algorithm, its window size is smaller than that of token bucket rate limiter, which is a capacity/rate. And move backward one time window at a time. Other rate limiter principles are similar to token bucket.
3. Concurrent rate limiter： Strictly limit the concurrent requests to N. Each time when there is a new request, it will check whether the number of concurrent requests is greater than N. If it is less than N, it is allowed to pass through, and the count is increased by 1. When the requests call ends, the signal is released (count minus 1).
4. Leaky bucket rate limiter: In contrast with token bucket algorithm, the leaky bucket algorithm can help to smooths the burst of requests and only allows a pre-defined N number of requests. This limiter can force the output flow at a constant rate of N. It is based on a leaky bucket model, the leaky water quantity is time interval*rate. if the leaky water quantity is greater than the number of has used (represented by key_bucket_count), then clear the bucket, that is, set the key_bucket_count to 0. Otherwise, set key_bucket_count minus the leaky water quantity. If the number (requests + key_bucket_count ) is less than the capacity, then allow the requests passing through.

Let's understand the functionality of callback() by reading concurrent rate limiter code, and understand the usage of getKeys() through reading the Lua script of token rate limiter and slide window rate limiter.

callback() used in Concurrent requests limiter#

The getKeys() method of ConcurrentRateLimiterAlgorithm overrides the template method in AbstractRateLimiterAlgorithm ：

    @Override    public List<String> getKeys(final String id) {        String tokenKey = getKeyName() + ".{" + id + "}.tokens";        String requestKey = UUIDUtils.getInstance().generateShortUuid();        return Arrays.asList(tokenKey, requestKey);    }

The second element, requestKey is a long type and non-duplicate value (generated by a distributed ID generator，it is incremented and smaller than the current time Epochsecond value). The corresponding Lua script in concurrent_request_rate_limiter.lua:

local key = KEYS[1]
local capacity = tonumber(ARGV[2])local timestamp = tonumber(ARGV[3])local id = KEYS[2]

Here id is requestKey generated by getKeys() method, it is an uuid(unique value). Subsequent process is as follows:

local count = redis.call("zcard", key)local allowed = 0
if count < capacity then  redis.call("zadd", key, timestamp, id)  allowed = 1  count = count + 1endreturn { allowed, count }

First, using zcard command to obtain the cardinality of the sorted set, and set count equals the cardinality , if the cardinality is less than the capacity, we will add a new member id (it is an uuid) to the sorted set, with the score of current time(in seconds) . then count =count+1, the cardinality is also incremented by 1 in reality.

All of the code above is executed in Redis as an atomic transaction. If there are a large number of concurrent requests from the same key( such as ip) , the cardinality of the sorted set of this key will increasing sharply, when then capacity limit is exceeded, the service will be denied, that is allowed =0。

In concurrent requests limiter, It is required to release the semaphore when the request is completed. However, it is not included in Lua script.

Let's see the callback function of ConcurrentRateLimiterAlgorithm：

    @Override    @SuppressWarnings("unchecked")    public void callback(final RedisScript<?> script, final List<String> keys, final List<String> scriptArgs) {        Singleton.INST.get(ReactiveRedisTemplate.class).opsForZSet().remove(keys.get(0), keys.get(1)).subscribe();    }

Here gives asynchronous subscription, using ReactiveRedisTemplate to delete the elements (key,id) in Redis store. That is once the request operation ends, the semaphore will be released. This remove operation cannot be executed in Lua script. This is just what design intention of callback in RateLimiterAlgorithm SPI .

getKeys() used in token bucket rate limiter#

Following is the corresponding Lua code:

local tokens_key = KEYS[1]local timestamp_key = KEYS[2]

Here we omit the code that get the parameters of rate ,capacity, etc.

local fill_time = capacity/ratelocal ttl = math.floor(fill_time*2)

The window size variable(ttl) is approximately two times of capacity/rate.

local last_tokens = tonumber(redis.call("get", tokens_key))if last_tokens == nil then  last_tokens = capacityend

Get last_tokens from the sorted set, if it not exist, then last_tokens equals capacity.

local last_refreshed = tonumber(redis.call("get", timestamp_key))if last_refreshed == nil then  last_refreshed = 0end

Get the last refreshed time by the key =timestamp_key from the sorted set, and default 0.

local delta = math.max(0, now-last_refreshed)local filled_tokens = math.min(capacity, last_tokens+(delta*rate))local allowed = filled_tokens >= requestedlocal allowed_num = 0if allowed then  new_tokens = filled_tokens - requested  allowed_num = 1end

The filled_tokens is produced evenly by time interval * rate，if the number of tokens greater than requests, then allowed=1, and update new_tokens.

redis.call("setex", tokens_key, ttl, new_tokens)redis.call("setex", timestamp_key, ttl, now)
return { allowed_num, new_tokens }

Here now is current time parameters passed in, set tokens_key to hold the string new_tokens and settokens_key to timeout after ttl of seconds. Set timestamp_key to hold the string value now, and expires after ttl seconds.

getKeys() used in sliding window rate limiter#

The getKeys() in SlidingWindowRateLimiterAlgorithm also overrides the parent class, and the code is consistent with the method in ConcurrentRateLimiterAlgorithm

Following is the Lua code of slide window rate limiter, the receiving of other parameters is omitted.

local timestamp_key = KEYS[2]...... local window_size = tonumber(capacity / rate)local window_time = 1

Here set the window_size to capacity/rate.

local last_requested = 0local exists_key = redis.call('exists', tokens_key)if (exists_key == 1) then    last_requested = redis.call('zcard', tokens_key)end

Obtain the cardinality(last_requested) of the tokens_key in the sorted set.

local remain_request = capacity - last_requestedlocal allowed_num = 0if (last_requested < capacity) then    allowed_num = 1    redis.call('zadd', tokens_key, now, timestamp_key)end

Calculate remaining available remain_request equals capacity minus last_requested . If last_requested less than capacity ,then allow current requests passing through，add element in the sorted set with (key=timestamp_key, value=now) .

redis.call('zremrangebyscore', tokens_key, 0, now - window_size / window_time)redis.call('expire', tokens_key, window_size)
return { allowed_num, remain_request }

Previously has set window_time=1, using zremrangebyscore command of Redis to remove all the elements in the sorted set stored at tokens_key with a score in [0,now - window_size / window_time] , that is, move the window a window size. Set the expire time of tokens_key to window_size.

In the template method getKeys(final String id) of AbstractRateLimiterAlgorithm，the second key ( represented y secondKey) is a fixed string which concat the input parameter{id}. As we can see from the above three algorithm codes, in the token bucket algorithm, secondKey will be updated to the latest time in the Lua code, so it doesn't matter what value is passed in. In the concurrent rate limiter, secondKey will be used as the key to remove Redis data in the java callback method. In the sliding window algorithm, the secondKey will be added to the sorted set as the key of a new element, and will be removed during window sliding.

That's all, when in a new rate limiter algorithm, the getKeys(final String id)method should be carefully designed according to the logic of the algorithm.

How to use RateLimiter SPI#

The three parameters in doExecute() method of RateLimiter plugin， exchange is an web request， chain is the execution chain of the plugins，selector is the selection parameters，rule is the policies or rules of rate limiter setting in the system.

protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {    //get  the `RateLimiterHandle` parameters from cache     RateLimiterHandle limiterHandle = RatelimiterRuleHandleCache.getInstance()        .obtainHandle(CacheKeyUtils.INST.getKey(rule));    //find the resolver name     String resolverKey = Optional.ofNullable(limiterHandle.getKeyResolverName())        .flatMap(name -> Optional.of("-" + RateLimiterKeyResolverFactory.newInstance(name).resolve(exchange)))        .orElse("");    return redisRateLimiter.isAllowed(rule.getId() + resolverKey, limiterHandle)        .flatMap(response -> {            if (!response.isAllowed()) {                exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);                Object error = ShenyuResultWrap.error(ShenyuResultEnum.TOO_MANY_REQUESTS.getCode(), ShenyuResultEnum.TOO_MANY_REQUESTS.getMsg(), null);                return WebFluxResultUtils.result(exchange, error);            }            return chain.execute(exchange);        });}

1. Firstly get the RateLimiterHandle parameters from cache.

2. Obtains the corresponding Key resolver by RateLimiterHandle instance.

3. Reactively executes isAllowed() method of RedisRateLimiter.

4. If not allowed, error handling is performed.

5. If the request is allowed, dispatch it to the next process of execution chain.

Summary#

RateLimiter plugin is based on Spring WebFlux,and with Apache Shen SPI, with Redis and Lua script to responsible for the critical algorithm and logic process, make it with characteristics of high concurrency and elastic. As for the RateLimiter SPI.

1. RateLimiter SPI provides two SPI interface, with interface oriented design and various design patterns, it's easy to develop new rate limiter algorithm and key resolver rule.
2. RateLimiterAlgorithm SPI supplies four rate limiter algorithms, token bucket,concurrency rate limiter, leaky bucket and sliding window rate limiter. When designing rate limiter algorithm, the KEY generation need to be carefully designed according to the algorithm characteristic. Using Lua script to realize the logic of the algorithm, and design callback() method for asynchronous processing when needed.
3. Reactive programming, simple and efficient implementation.

Apache ShenYu is an asynchronous, high-performance, cross-language, responsive API gateway.

background#

Recently,when I read the source code of open source project Apache Shenyu API gateway,I find and many core components of the gateway are loaded with the SPI module. Here I will analyzes the source code of SPI module in Shenyu gateway.

what is SPI#

SPI means 'Service Provider Interface', which is a dynamic Service discovery mechanism. We can dynamically load the implementation class of the Interface based on the runtime of the Interface (that is, a development mode of Interface programming + strategy pattern + configuration file) with it. The most common is the built-in database Driver interface 'java.sql.Driver' in JDK. Different vendors can implement this interface differently. For example, 'MySQL' ('com.mysql.jdbc.Driver' in the 'MySQL' Driver package),' PostgreSQL' ('org.postgresql.driver' in the 'PostgreSQL' Driver package), etc.

The JDK's built-in 'SPI' can be used as follows:

• In the 'META-INF/services' directory of the classpath, create a file named with the fully qualified name of the interface (essentially a 'properties' file) whose implementation classes you want the SPI loader to load , for example if you want to load the SQL driver implementation classes mentioned above then create a file named 'java.sql.Driver' since those classes implement the 'java.sql.driver' interface.

• In this file we can add entries for all the specific implementations of the interface . For example for the above driver class scenario we would add entries as shown in below code snippet in the file META-INF/services/java.sql.Driver

# content of file META-INF/services/java.sql.Drivercom.mysql.jdbc.Driverorg.postgresql.Driver

• Finally load the file with 'java.util.ServiceLoader' to instantiate the corresponding implementation class of the interface

ServiceLoader<Driver> loader = ServiceLoader.load(Driver.class)

The underlying implementation involves classloading, the parent delegate model, and so on, which I won't expand here. Based on this design idea, many mainstream frameworks self-implemented a set of 'SPI' extension, such as 'Dubbo SPI' extension module, which would read the 'META-INF/services/dubbo' directory file content in the classppath for class loading. The 'shenyu-spi' module also follows this similar design idea.

source code of shenyu-spi#

The 'shenyu-spi' module is very concise, and the code structure is as follows:

- shenyu-spi[module]  - org.apache.shenyu.spi[package]    -- ExtensionFactory    -- ExtensionLoader    -- Join    -- SPI    -- SpiExtensionFactory

这些类功能如下：

• 'ExtensionFactory' : 'SPI' loader Factory, used to load an 'ExtensionLoader' instance based on the 'SPI' mechanism and to obtain the default 'SPI' identity implementation based on the 'ExtensionLoader' instance
• 'SpiExtensionFactory' : is an implementation of 'ExtensionFactory'
• 'SPI' : identification annotation, used to identify 'SPI', used on the interface
• 'Join' : identification annotation, used on the implementation class, used to identify the class joining the SPI system
• 'ExtensionLoader' : 'SPI' loader, analogous to 'java.util.ServiceLoader', used to load the implementation class of the interface in 'SPI'

@SPI#

org.apache.shenyu.spi.SPI is an identification annotation which is used for identifying an interface as a 'SPI' interface.That is, only interfaces that use '@SPI' annotation can be loaded by 'shenyu-spi'. The class's annotation describes the implementation of Apache Dubbo, a reference to all the SPI systems (which makes sense, since the SPI extension is already a mature scheme with much the same implementation). This annotation has only one method:

@Documented@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.TYPE)public @interface SPI {
    /**     * Value string.     *     * @return the string     */    String value() default "";}

The unique 'value()' method is used to specify the default 'SPI' implementation (optional), as will be shown later when analyzing 'ExtensionLoader'.

@Join#

org.apache.shenyu.spi.Join is an identification annotation too. When this annotation is used on a class it specifies that the class contains 'SPI' implementation and to indicate that the class is added to the SPI system and can be loaded by ExtensionLoader. This annotation has two methods:

@Documented@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.TYPE)public @interface Join {        /**     * It will be sorted according to the current serial number..     * @return int.     */    int order() default 0;
    /**     * Indicates that the object joined by @Join is a singleton,     * otherwise a completely new instance is created each time.     * @return true or false.     */    boolean isSingleton() default true;}

The unique 'order()' method is used to specify the specific sequence number. If a single interface that annotated with '@SPI' has multiple implementation classes that annotated with '@Join', the sequence number determines the order of these implementation class instances (the smaller one comes first).

The isSingleton() method indicates whether the class that the implementation class is a singleton class or not. That is if it is a singleton class it will be instantiated only once else it will create a new instance everytime .

ExtensionLoader#

'ExtensionLoader' is the core of the 'SPI' module. Look at it's attributes first:

public final class ExtensionLoader<T> {        // SLF4J日志句柄    private static final Logger LOG = LoggerFactory.getLogger(ExtensionLoader.class);        // SPI配置文件基于类路径下的相对目录    private static final String SHENYU_DIRECTORY = "META-INF/shenyu/";        // @SPI标识接口类型 -> ExtensionLoader实例的缓存 => 注意这个是一个全局的静态变量    private static final Map<Class<?>, ExtensionLoader<?>> LOADERS = new ConcurrentHashMap<>();        // 当前@SPI标识接口类型    private final Class<T> clazz;        // 类加载器实例    private final ClassLoader classLoader;        // 当前ExtensionLoader缓存的已加载的实现类信息，使用值持有器包装，是一个HashMap，映射关系：实现类别名 -> 实现类信息    private final Holder<Map<String, ClassEntity>> cachedClasses = new Holder<>();        // 当前ExtensionLoader缓存的已加载的实现类实例的值包装器，使用值持有器包装，映射关系：实现类别名 -> 值持有器包装的实现类实体    private final Map<String, Holder<Object>> cachedInstances = new ConcurrentHashMap<>();        // 当前ExtensionLoader缓存的已加载的实现类实例，使用值持有器包装，映射关系：实现类类型 -> 实现类实体    private final Map<Class<?>, Object> joinInstances = new ConcurrentHashMap<>();        // 缓存默认名称，来源于@SPI注解的value()方法非空白返回值，用于加载默认的接口实现    private String cachedDefaultName;        // Holder比较器，按照Holder的order降序，也就是顺序号小的排在前面    private final Comparator<Holder<Object>> holderComparator = Comparator.comparing(Holder::getOrder);        // ClassEntity比较器，按照ClassEntity的order降序，也就是顺序号小的排在前面    private final Comparator<ClassEntity> classEntityComparator = Comparator.comparing(ClassEntity::getOrder);        // 暂时省略其他代码
    // 值持有器，简单VO，用来存储泛型值和值加载顺序    public static class Holder<T> {                // 这里的值引用是volatile修饰，便于某线程更变另一线程马上读到最新的值        private volatile T value;                private Integer order;
        private boolean isSingleton;                // 省略setter和getter代码    }        // 类实体，主要存放加载的实现类的信息    static final class ClassEntity {                // 名称，这里是指SPI实现类的别名，不是类名        private String name;                // 加载顺序号        private Integer order;
        private Boolean isSingleton;                // SPI实现类        private Class<?> clazz;
        private ClassEntity(final String name, final Integer order, final Class<?> clazz, final boolean isSingleton) {            this.name = name;            this.order = order;            this.clazz = clazz;            this.isSingleton = isSingleton;        }                // 省略setter和getter代码    }}

After analyzing the attributes, it is not difficult to find the following points:

• 'ExtensionLoader' There will be a global static cache 'LOADERS' to cache already created instances of 'ExtensionLoader' to prevent the performance overhead of repeated creation
• Each '@SPI' interface that is loaded using 'ExtensionLoader' generates a new instance of 'ExtensionLoader'
• '@SPI' interfaces that have multiple implementations are eventually acquired in order

Then look at it's constructors and static factory methods:

// 私有构造函数，需要入参为@SPI标识的接口类型和类加载器实例private ExtensionLoader(final Class<T> clazz, final ClassLoader cl) {    // 成员变量clazz赋值    this.clazz = clazz;    // 成员变量classLoader赋值    this.classLoader = cl;    // 这里对于非ExtensionFactory接口类型会懒加载一个用于加载ExtensionFactory的ExtensionLoader    if (!Objects.equals(clazz, ExtensionFactory.class)) {        ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getExtensionClassesEntity();    }}
// 实例化getExtensionLoader，静态工厂方法，需要入参为@SPI标识的接口类型和类加载器实例public static <T> ExtensionLoader<T> getExtensionLoader(final Class<T> clazz, final ClassLoader cl) {    // 前缀校验，接口类型必须非空并且必须存在@SPI注解，否则抛出异常中断    Objects.requireNonNull(clazz, "extension clazz is null");    if (!clazz.isInterface()) {        throw new IllegalArgumentException("extension clazz (" + clazz + ") is not interface!");    }    if (!clazz.isAnnotationPresent(SPI.class)) {        throw new IllegalArgumentException("extension clazz (" + clazz + ") without @" + SPI.class + " Annotation");    }    // 从缓存LOADERS中加载ExtensionLoader实例，不存在则创建，典型的懒加载模式    ExtensionLoader<T> extensionLoader = (ExtensionLoader<T>) LOADERS.get(clazz);    if (Objects.nonNull(extensionLoader)) {        return extensionLoader;    }    LOADERS.putIfAbsent(clazz, new ExtensionLoader<>(clazz, cl));    return (ExtensionLoader<T>) LOADERS.get(clazz);}
// 实例化getExtensionLoader，静态工厂方法，需要入参为@SPI标识的接口类型，使用ExtensionLoader类的类加载器public static <T> ExtensionLoader<T> getExtensionLoader(final Class<T> clazz) {    return getExtensionLoader(clazz, ExtensionLoader.class.getClassLoader());}

'ExtensionLoader' uses a private constructor, static factory methods, and lazy loading mode. Class loading is not triggered after initializing 'ExtensionLoader'. The actual scanning and loading is delayed until the 'getJoin' series methods are called, where the code is swept and all the method call chains that implement class information are loaded:

// 加载所有扩展类信息，这里采用了DCL（双重锁校验）防止并发加载private Map<String, ClassEntity> getExtensionClassesEntity() {    // 缓存不存在    Map<String, ClassEntity> classes = cachedClasses.getValue();    if (Objects.isNull(classes)) {        // 加锁后再检查一次缓存        synchronized (cachedClasses) {            classes = cachedClasses.getValue();            if (Objects.isNull(classes)) {                // 最终确认缓存不存在，则进行加载，并且标记顺序号为0                classes = loadExtensionClass();                cachedClasses.setValue(classes);                cachedClasses.setOrder(0);            }        }    }    return classes;}
// 加载当前ExtensionLoader中clazz的所有SPI系统内的实现类private Map<String, ClassEntity> loadExtensionClass() {    SPI annotation = clazz.getAnnotation(SPI.class);    if (Objects.nonNull(annotation)) {        // 这里就是前面提到，如果@SPI注解的value()方法非空白返回值会作为默认实现的别名        // 也就是如果只使用了@SPI，那么就无法获取默认实现        // 如果使用了@SPI("foo")，可以通过别名foo去映射和获取默认实现        String value = annotation.value();        if (StringUtils.isNotBlank(value)) {            cachedDefaultName = value;        }    }    // 初始化一个Hashmap容器用于存储加载的实现类信息，这个变量会透传到下一个方法链    Map<String, ClassEntity> classes = new HashMap<>(16);    // 加载目录中的属性文件    loadDirectory(classes);    return classes;}
// 加载目录中的属性文件，并且加载文件中的实现类，目标目录：META-INF/shenyu/private void loadDirectory(final Map<String, ClassEntity> classes) {    // 文件名 => META-INF/shenyu/$className    String fileName = SHENYU_DIRECTORY + clazz.getName();    try {        // 这里使用类加载器加载文件资源，如果传入的类加载器为空会使用系统类加载器        Enumeration<URL> urls = Objects.nonNull(this.classLoader) ? classLoader.getResources(fileName)                : ClassLoader.getSystemResources(fileName);        // 遍历解析的文件URL集合        if (Objects.nonNull(urls)) {            while (urls.hasMoreElements()) {                URL url = urls.nextElement();                // 通过文件URL加载资源                loadResources(classes, url);            }        }    } catch (IOException t) {        LOG.error("load extension class error {}", fileName, t);    }}
// 加载文件资源，解析文件并且加载实现类存储到classes中private void loadResources(final Map<String, ClassEntity> classes, final URL url) throws IOException {    // 读取URL文件资源，加载到Properties中，每行格式为name=classPath    try (InputStream inputStream = url.openStream()) {        Properties properties = new Properties();        properties.load(inputStream);        properties.forEach((k, v) -> {            String name = (String) k;            String classPath = (String) v;            if (StringUtils.isNotBlank(name) && StringUtils.isNotBlank(classPath)) {                try {                    // 基于name和classPath进行类加载                    loadClass(classes, name, classPath);                } catch (ClassNotFoundException e) {                    throw new IllegalStateException("load extension resources error", e);                }            }        });    } catch (IOException e) {        throw new IllegalStateException("load extension resources error", e);    }}
// 基于name（别名）和classPath（类全限定名称）进行类加载private void loadClass(final Map<String, ClassEntity> classes,                        final String name, final String classPath) throws ClassNotFoundException {    // 类初始化，并且确定实现类必须是当前@SPI注解标识接口的子类    Class<?> subClass = Objects.nonNull(this.classLoader) ? Class.forName(classPath, true, this.classLoader) : Class.forName(classPath);    if (!clazz.isAssignableFrom(subClass)) {        throw new IllegalStateException("load extension resources error," + subClass + " subtype is not of " + clazz);    }    // 实现类必须存在注解@Join    if (!subClass.isAnnotationPresent(Join.class)) {        throw new IllegalStateException("load extension resources error," + subClass + " without @" + Join.class + " annotation");    }    // 如果缓存中不存在同样别名的实现类才进行缓存，已经存在则校验旧的类型和当前实现类型是否一致    ClassEntity oldClassEntity = classes.get(name);    if (Objects.isNull(oldClassEntity)) {        // 创建类信息实体保存别名、顺序号和实现类并且缓存，映射关系：别名 -> 类信息实体        Join joinAnnotation = subClass.getAnnotation(Join.class);        ClassEntity classEntity = new ClassEntity(name, joinAnnotation.order(), subClass);        classes.put(name, classEntity);    } else if (!Objects.equals(oldClassEntity.getClazz(), subClass)) {        throw new IllegalStateException("load extension resources error,Duplicate class " + clazz.getName() + " name "                + name + " on " + oldClassEntity.getClazz().getName() + " or " + subClass.getName());    }}

Processing with the chain of method 'getExtensionClassesEntity - > loadExtensionClass - > loadDirectory - > loadResources - > LoadClass',it will create a mapping of 'alias' to 'implementation class information' for subsequent instantiations, as shown in the 'getJoin()' method:

// 基于别名获取实现类实例public T getJoin(final String name) {    // 别名必须为非空白字符串    if (StringUtils.isBlank(name)) {        throw new NullPointerException("get join name is null");    }    // 这里也使用DCL去cachedInstances缓存中取别名对应的值持有器，值持有器为空则创建    Holder<Object> objectHolder = cachedInstances.get(name);    if (Objects.isNull(objectHolder)) {        cachedInstances.putIfAbsent(name, new Holder<>());        objectHolder = cachedInstances.get(name);    }    Object value = objectHolder.getValue();    if (Objects.isNull(value)) {        synchronized (cachedInstances) {            // 加锁后再次判断值持有器中的值是否存在，不存在的时候则进行实现类实例化            value = objectHolder.getValue();            if (Objects.isNull(value)) {                Holder<T> pair = createExtension(name);                value = pair.getValue();                int order = pair.getOrder();                // 实例化完成后更新值持有器缓存                objectHolder.setValue(value);                objectHolder.setOrder(order);            }        }    }    return (T) value;}
// 基于别名搜索已经加载的实现类信息，并且实例化对应的实现类进行值包装private Holder<T> createExtension(final String name) {    // 加载该@SPI标识接口的所有实现类信息并且获取对应别名的实现类信息    ClassEntity classEntity = getExtensionClassesEntity().get(name);    if (Objects.isNull(classEntity)) {        throw new IllegalArgumentException("name is error");    }    Class<?> aClass = classEntity.getClazz();    // 如果实现类实例缓存中已经存在，则直接封装为值包装器返回，否则进行实例化    Object o = joinInstances.get(aClass);    if (Objects.isNull(o)) {        try {            // 反射实例化并且缓存该实现类实例            joinInstances.putIfAbsent(aClass, aClass.newInstance());            o = joinInstances.get(aClass);        } catch (InstantiationException | IllegalAccessException e) {            throw new IllegalStateException("Extension instance(name: " + name + ", class: "                    + aClass + ")  could not be instantiated: " + e.getMessage(), e);                    }    }    Holder<T> objectHolder = new Holder<>();    objectHolder.setOrder(classEntity.getOrder());    objectHolder.setValue((T) o);    return objectHolder;}

As you can see from the 'createExtension()' method, we end up using reflection to instantiate the implementation class. The reflection method 'newInstance()' requires that the class must provide a no-argument constructor because of an implicit convention: The 'SPI' implementation class must provide a no-argument constructor or the instantiation will fail. The rest methods, such as 'getDefaultJoin()' and 'getJoins()' are uncomplicated extensions of 'getJoin()', so we won't analyze them here. In addition, the 'getJoin()' method uses a multilevel cache:

• 'cachedInstances' : Search for the corresponding implementation class instance by alias
• 'joinInstances' : If the alias lookup fails, load all the implementation class information, locate the implementation class type by the alias, and update the' cachedInstances' cache by either finding the implementation class type or creating and caching the implementation class instance

This completes the source code analysis of 'ExtensionLoader'. Here's another example of an 'ExtensionLoader' instance member property memory layout diagram to help you understand:

ExtensionFactory#

'ExtensionFactory' is the factory interface inside the factory pattern, which defines a method to get an instance of the 'SPI' implementation (the default implementation, or the only implementation) :

@SPI("spi")public interface ExtensionFactory {
    /**     * Gets Extension.     *     * @param <T>   the type parameter     * @param key   此参数暂时没有使用，猜测是预留用于映射@SPI的value()     * @param clazz @SPI标识的接口类型     * @return the extension     */    <T> T getExtension(String key, Class<T> clazz);}

Let's look the class 'SpiExtensionFactory' :

@Joinpublic class SpiExtensionFactory implements ExtensionFactory {
    @Override    public <T> T getExtension(final String key, final Class<T> clazz) {        return Optional.ofNullable(clazz)   // 入参clazz非空                .filter(Class::isInterface)  // 入参clazz必须是接口                .filter(cls -> cls.isAnnotationPresent(SPI.class))  // 入参clazz必须被@SPI标识                .map(ExtensionLoader::getExtensionLoader)  // 基于clazz这个接口类型实例化ExtensionLoader                .map(ExtensionLoader::getDefaultJoin)  // 获取该@SPI标识接口的默认实现，不存在则返回NULL                .orElse(null);    }}

It's worth noting here that the 'ExtensionFactory' itself is part of the 'SPI' system. So when using 'ExtensionFactory' you can instantiate it directly:

ExtensionFactory extensionFactory = new SpiExtensionFactory();

It can also be loaded based on an 'ExtensionLoader':

# the content of META-INF/services/shenyu/org.apache.shenyu.spi.ExtensionFactoryspi=org.apache.shenyu.spi.SpiExtensionFactory
# then load it with ExtensionLoaderExtensionFactory extensionFactory = ExtensionLoader.getExtensionLoader(ExtensionFactory.class).getDefaultJoin();

Once you have an 'ExtensionFactory' instance, you can load an instance of its default implementation class based on the '@SPI' interface.

小结#

The 'SPI' extension framework based on the design idea of 'Java' native 'SPI' has the characteristics of loose coupling, high usability and high scalability, a loading instance cache system, concurrency security and other features to fill some defects of 'SPI' in the native 'JDK', Shenyu SPI module is the same. Base on this powerful 'SPI' module, other modules in 'Shenyu' such as the 'Plugin' module can be configured quickly and pluggable, making it easier to load a newly developed Plugin instance.