Dubbo3高级特性「框架与服务」服务并发控制及集群负载均衡的实践指南（含扩展SPI）
Posted 洛神灬殇
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Dubbo3高级特性「框架与服务」服务并发控制及集群负载均衡的实践指南（含扩展SPI）相关的知识，希望对你有一定的参考价值。
Dubbo3中的并发控制
XML方式配置
限制类的线程隔离控制（服务端）
限制com.xxx.ApiService的每个方法，服务器端并发执行（或占用线程池线程数）不能超过 10 个：
<dubbo:service interface="com.xxx.ApiService" executes="10" />
Annotation方式配置
限制类的线程隔离控制（服务端）
限制com.xxx.ApiService的每个方法，服务器端并发执行（或占用线程池线程数）不能超过 10 个：
@DubboService(version = "1.0.0",executes=10)
public class DefaultRpcOrderProcessApi implements RpcOrderProcessApi
限制方法的线程隔离控制（服务端）
限制com.xxx.ApiService的sayHello 方法，服务器端并发执行（或占用线程池线程数）不能超过 10 个：
<dubbo:service interface="com.xxx.ApiService">
<dubbo:method name="sayHello" executes="10" />
</dubbo:service>
Annotation方式配置
限制方法的线程隔离控制（服务端）
限制com.xxx.ApiService的sayHello 方法，服务器端并发执行（或占用线程池线程数）不能超过 10 个：
@DubboService(version = "1.0.0",
methods =
@Method(name = "sayHello", executes= 10, retries = 0))
public class DefaultRpcOrderProcessApi implements RpcOrderProcessApi
XML方式配置
限制类的线程隔离控制（消费端）
限制 com.xxx.ApiService的每个方法，每客户端并发执行（或占用连接的请求数）不能超过 10 个：
<dubbo:reference interface="com.xxx.ApiService" actives="10" />
Annotation方式配置
限制类的线程隔离控制（消费端）
限制 com.xxx.ApiService的每个方法，每客户端并发执行（或占用连接的请求数）不能超过 10 个：
@DubboReference(version = "1.0.0",actives = 10)
RpcShopCarProcessApi rpcShopCarProcessApi;
XML方式配置
限制方法的线程隔离控制（消费端）
限制 com.xxx.ApiService 的 sayHello 方法，每客户端并发执行（或占用连接的请求数）不能超过 10 个：
<dubbo:reference interface="com.xxx.ApiService">
<dubbo:method name="sayHello" actives="10" />
</dubbo:service>
Annotation方式配置
限制方法的线程隔离控制（消费端）
限制 com.xxx.ApiService 的 sayHello 方法，每客户端并发执行（或占用连接的请求数）不能超过 10 个：
@DubboReference(version = "1.0.0",methods = @Method(name = "sayHello", actives= 10, retries = 0))
RpcShopCarProcessApi rpcShopCarProcessApi;```
集群负载均衡
在集群负载均衡时，Dubbo 提供了多种均衡策略，缺省为 random 随机调用。配置服务的客户端的 loadbalance 属性为leastactive，此 Loadbalance 会调用并发数最小的 Provider（Consumer端并发数）。
负载均衡策略
目前 Dubbo 内置了如下负载均衡算法，用户可直接配置使用：
| 算法 | 特性 | 备注 |
|---|---|---|
| RandomLoadBalance | 加权随机 | 默认算法，默认权重相同 |
| RoundRobinLoadBalance | 加权轮询 | 借鉴于 nginx 的平滑加权轮询算法，默认权重相同 |
| LeastActiveLoadBalance | 最少活跃优先 + 加权随机 | 背后是能者多劳的思想 |
| ShortestResponseLoadBalance | 最短响应优先 + 加权随机 | 更加关注响应速度 |
| ConsistentHashLoadBalance | 一致性 Hash | 确定的入参，确定的提供者，适用于有状态请求 |
LoadbalanceRules的选择类型
public interface LoadbalanceRules
/**
* This class select one provider from multiple providers randomly.
**/
String RANDOM = "random";
/**
* Round robin load balance.
**/
String ROUND_ROBIN = "roundrobin";
/**
* Filter the number of invokers with the least number of active calls and count the weights and quantities of these invokers.
**/
String LEAST_ACTIVE = "leastactive";
/**
* Consistent Hash, requests with the same parameters are always sent to the same provider.
**/
String CONSISTENT_HASH = "consistenthash";
/**
* Filter the number of invokers with the shortest response time of success calls and count the weights and quantities of these invokers.
**/
String SHORTEST_RESPONSE = "shortestresponse";
String EMPTY = "";
Random
加权随机，按权重设置随机概率。在一个截面上碰撞的概率高，但调用量越大分布越均匀，而且按概率使用权重后也比较均匀，有利于动态调整提供者权重。
- 缺点：存在慢的提供者累积请求的问题，比如：第二台机器很慢，但没挂，当请求调到第二台时就卡在那，久而久之，所有请求都卡在调到第二台上。
实现的源码如下：
public class RandomLoadBalance extends AbstractLoadBalance
public static final String NAME = "random";
/**
* Select one invoker between a list using a random criteria
* @param invokers List of possible invokers
* @param url URL
* @param invocation Invocation
* @param <T>
* @return The selected invoker
*/
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)
// Number of invokers
int length = invokers.size();
if (!needWeightLoadBalance(invokers,invocation))
return invokers.get(ThreadLocalRandom.current().nextInt(length));
// Every invoker has the same weight?
boolean sameWeight = true;
// the maxWeight of every invokers, the minWeight = 0 or the maxWeight of the last invoker
int[] weights = new int[length];
// The sum of weights
int totalWeight = 0;
for (int i = 0; i < length; i++)
int weight = getWeight(invokers.get(i), invocation);
// Sum
totalWeight += weight;
// save for later use
weights[i] = totalWeight;
if (sameWeight && totalWeight != weight * (i + 1))
sameWeight = false;
if (totalWeight > 0 && !sameWeight)
// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
int offset = ThreadLocalRandom.current().nextInt(totalWeight);
// Return a invoker based on the random value.
for (int i = 0; i < length; i++)
if (offset < weights[i])
return invokers.get(i);
// If all invokers have the same weight value or totalWeight=0, return evenly.
return invokers.get(ThreadLocalRandom.current().nextInt(length));
private <T> boolean needWeightLoadBalance(List<Invoker<T>> invokers, Invocation invocation)
Invoker invoker = invokers.get(0);
URL invokerUrl = invoker.getUrl();
// Multiple registry scenario, load balance among multiple registries.
if (REGISTRY_SERVICE_REFERENCE_PATH.equals(invokerUrl.getServiceInterface()))
String weight = invokerUrl.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY);
if (StringUtils.isNotEmpty(weight))
return true;
else
String weight = invokerUrl.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY);
if (StringUtils.isNotEmpty(weight))
return true;
else
String timeStamp = invoker.getUrl().getParameter(TIMESTAMP_KEY);
if (StringUtils.isNotEmpty(timeStamp))
return true;
return false;
当我们使用它的时候，只需要进行选择loadBalance值作为“random”（ public static final String NAME = “random”;）
RoundRobin
加权轮询，按公约后的权重设置轮询比率，循环调用节点
- 缺点：同样存在慢的提供者累积请求的问题。
源码如下：
public class RoundRobinLoadBalance extends AbstractLoadBalance
public static final String NAME = "roundrobin";
private static final int RECYCLE_PERIOD = 60000;
protected static class WeightedRoundRobin
private int weight;
private AtomicLong current = new AtomicLong(0);
private long lastUpdate;
public int getWeight()
return weight;
public void setWeight(int weight)
this.weight = weight;
current.set(0);
public long increaseCurrent()
return current.addAndGet(weight);
public void sel(int total)
current.addAndGet(-1 * total);
public long getLastUpdate()
return lastUpdate;
public void setLastUpdate(long lastUpdate)
this.lastUpdate = lastUpdate;
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
/**
* get invoker addr list cached for specified invocation
* <p>
* <b>for unit test only</b>
*
* @param invokers
* @param invocation
* @return
*/
protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation)
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
if (map != null)
return map.keySet();
return null;
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
int totalWeight = 0;
long maxCurrent = Long.MIN_VALUE;
long now = System.currentTimeMillis();
Invoker<T> selectedInvoker = null;
WeightedRoundRobin selectedWRR = null;
for (Invoker<T> invoker : invokers)
String identifyString = invoker.getUrl().toIdentityString();
int weight = getWeight(invoker, invocation);
WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k ->
WeightedRoundRobin wrr = new WeightedRoundRobin();
wrr.setWeight(weight);
return wrr;
);
if (weight != weightedRoundRobin.getWeight())
//weight changed
weightedRoundRobin.setWeight(weight);
long cur = weightedRoundRobin.increaseCurrent();
weightedRoundRobin.setLastUpdate(now);
if (cur > maxCurrent)
maxCurrent = cur;
selectedInvoker = invoker;
selectedWRR = weightedRoundRobin;
totalWeight += weight;
if (invokers.size() != map.size())
map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
if (selectedInvoker != null)
selectedWRR.sel(totalWeight);
return selectedInvoker;
// should not happen here
return invokers.get(0);
得出结论可以使用roundrobin进行指定（public static final String NAME = "roundrobin）
LeastActive
加权最少活跃调用优先，活跃数越低，越优先调用，相同活跃数的进行加权随机。活跃数指调用前后计数差（针对特定提供者：请求发送数 - 响应返回数），表特定提供者的任务堆积量，活跃数越低，代表该提供者处理能力越强。
- 使慢的提供者收到更少请求，因为越慢的提供者的调用前后计数差会越大；相对的，处理能力越强的节点，处理更多的请求。
public class LeastActiveLoadBalance extends AbstractLoadBalance
public static final String NAME = "leastactive";
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)
// Number of invokers
int length = invokers.size();
// The least active value of all invokers
int leastActive = -1;
// The number of invokers having the same least active value (leastActive)
int leastCount = 0;
// The index of invokers having the same least active value (leastActive)
int[] leastIndexes = new int[length];
// the weight of every invokers
int[] weights = new int[length];
// The sum of the warmup weights of all the least active invokers
int totalWeight = 0;
// The weight of the first least active invoker
int firstWeight = 0;
// Every least active invoker has the same weight value?
boolean sameWeight = true;
// Filter out all the least active invokers
for (int i = 0; i < length; i++)
Invoker<T> invoker = invokers.get(i);
// Get the active number of the invoker
int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
// Get the weight of the invoker's configuration. The default value is 100.
int afterWarmup = getWeight(invoker, invocation);
// save for later use
weights[i] = afterWarmup;
// If it is the first invoker or the active number of the invoker is less than the current least active number
if (leastActive == -1 || active < leastActive)
// Reset the active number of the current invoker to the least active number
leastActive = active;
// Reset the number of least active invokers
leastCount = 1;
// Put the first least active invoker first in leastIndexes
leastIndexes[0] = i;
// Reset totalWeight
totalWeight = afterWarmup;
// Record the weight the first least active invoker
firstWeight = afterWarmup;
// Each invoke has the same weight (only one invoker here)
sameWeight = true;
// If current invoker's active value equals with leaseActive, then accumulating.
else if (active == leastActive)
// Record the index of the least active invoker in leastIndexes order
leastIndexes[leastCount++] = i;
// Accumulate the total weight of the least active invoker
totalWeight += afterWarmup;
// If every invoker has the same weight?
if (sameWeight && afterWarmup != firstWeight)
sameWeight = false;
// Choose an invoker from all the 以上是关于Dubbo3高级特性「框架与服务」服务并发控制及集群负载均衡的实践指南（含扩展SPI）的主要内容，如果未能解决你的问题，请参考以下文章
Dubbo3高级特性「框架与服务」Dubbo3客户端和服务端的泛化调用机制体系
Dubbo3高级特性「框架与服务」自定义Dubbo服务容器及扩展容器实现分析
Dubbo3高级特性「框架与服务」框架与服务的异步调用实践以及开发模式
Dubbo3高级特性「框架与服务」在dubbo3中进行参数校验及自定义验证扩展机制