dubbo集羣和負載均衡
Cluster接口的做用是將RegistryDirectory目錄中的List 包裝爲Invoker接口，讓Invoker接口調用方不須要關注負載均衡的細節。java
@SPI(FailoverCluster.NAME)
public interface Cluster {
@Adaptive
<T> Invoker<T> join(Directory<T> directory) throws RpcException;
}
@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
@Adaptive("loadbalance")
<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}
複製代碼
Cluster SPI接口默認使用FailoverCluster做爲實現類，Failover就是當前服務節點調用發送網絡異常時，重試其它節點。算法
Cluster實現會使用LoadBalance來實現多個Invoker的動態選擇。默認使用隨機算法實現。服務器
/** * 失敗轉移，當出現失敗，重試其它服務器，一般用於讀操做，但重試會帶來更長延遲。 * * <a href="http://en.wikipedia.org/wiki/Failover">Failover</a> * * @author william.liangf */
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailoverClusterInvoker<T>(directory);
}
}
public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
private static final Logger logger = LoggerFactory
.getLogger(AbstractClusterInvoker.class);
protected final Directory<T> directory;
protected final boolean availablecheck;
private volatile boolean destroyed = false;
private volatile Invoker<T> stickyInvoker = null;
public AbstractClusterInvoker(Directory<T> directory) {
this(directory, directory.getUrl());
}
public AbstractClusterInvoker(Directory<T> directory, URL url) {
if (directory == null)
throw new IllegalArgumentException("service directory == null");
this.directory = directory ;
//sticky 須要檢測 avaliablecheck
this.availablecheck = url.getParameter(Constants.CLUSTER_AVAILABLE_CHECK_KEY, Constants.DEFAULT_CLUSTER_AVAILABLE_CHECK) ;
}
public Class<T> getInterface() {
return directory.getInterface();
}
public URL getUrl() {
return directory.getUrl();
}
public boolean isAvailable() {
Invoker<T> invoker = stickyInvoker;
if (invoker != null) {
return invoker.isAvailable();
}
return directory.isAvailable();
}
public void destroy() {
directory.destroy();
destroyed = true;
}
public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
//調用註冊目錄獲取可用invoker列表
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
//獲取配置好的LoadBalance實現，若是沒有配置使用默認的隨機算法LoadBalance實現類
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
//抽象類，之類提供實現
protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException;
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}
/** * 使用loadbalance選擇invoker.</br> * a)先lb選擇，若是在selected列表中 或者 不可用且作檢驗時，進入下一步(重選),不然直接返回</br> * b)重選驗證規則：selected > available .保證重選出的結果儘可能不在select中，而且是可用的 * * @param availablecheck 若是設置true，在選擇的時候先選invoker.available == true * @param selected 已選過的invoker.注意：輸入保證不重複 * */
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
String methodName = invocation == null ? "" : invocation.getMethodName();
//dubbo能夠基於Service方法級別配置調用親和性，將該方法請求都固定在一臺服務器節點上。默認關閉，通常狀況下用不到這個特性。
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName,Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY) ;
{
//ignore overloaded method
if ( stickyInvoker != null && !invokers.contains(stickyInvoker) ){
stickyInvoker = null;
}
//ignore cucurrent problem
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))){
if (availablecheck && stickyInvoker.isAvailable()){
return stickyInvoker;
}
}
}
Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
if (sticky){
stickyInvoker = invoker;
}
return invoker;
}
private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
if (invokers.size() == 1)
return invokers.get(0);
// 若是隻有兩個invoker，退化成輪循
if (invokers.size() == 2 && selected != null && selected.size() > 0) {
return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
}
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//若是 selected中包含（優先判斷） 或者 不可用&&availablecheck=true 則重試.
if( (selected != null && selected.contains(invoker))
||(!invoker.isAvailable() && getUrl()!=null && availablecheck)){
try{
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if(rinvoker != null){
invoker = rinvoker;
}else{
//看下第一次選的位置，若是不是最後，選+1位置.
int index = invokers.indexOf(invoker);
try{
//最後在避免碰撞
invoker = index <invokers.size()-1?invokers.get(index+1) :invoker;
}catch (Exception e) {
logger.warn(e.getMessage()+" may because invokers list dynamic change, ignore.",e);
}
}
}catch (Throwable t){
logger.error("clustor relselect fail reason is :"+t.getMessage() +" if can not slove ,you can set cluster.availablecheck=false in url",t);
}
}
return invoker;
}
/** * 重選，先從非selected的列表中選擇，沒有在從selected列表中選擇. * @param loadbalance * @param invocation * @param invokers * @param selected * @return * @throws RpcException */
private Invoker<T> reselect(LoadBalance loadbalance,Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected ,boolean availablecheck) throws RpcException {
//預先分配一個，這個列表是必定會用到的.
List<Invoker<T>> reselectInvokers = new ArrayList<Invoker<T>>(invokers.size()>1?(invokers.size()-1):invokers.size());
//先從非select中選
if( availablecheck ){ //選isAvailable 的非select
for(Invoker<T> invoker : invokers){
if(invoker.isAvailable()){
if(selected ==null || !selected.contains(invoker)){
reselectInvokers.add(invoker);
}
}
}
if(reselectInvokers.size()>0){
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}else{ //選所有非select
for(Invoker<T> invoker : invokers){
if(selected ==null || !selected.contains(invoker)){
reselectInvokers.add(invoker);
}
}
if(reselectInvokers.size()>0){
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
//最後從select中選可用的.
{
if(selected != null){
for(Invoker<T> invoker : selected){
if((invoker.isAvailable()) //優先選available
&& !reselectInvokers.contains(invoker)){
reselectInvokers.add(invoker);
}
}
}
if(reselectInvokers.size()>0){
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
return null;
}
protected void checkWheatherDestoried() {
if(destroyed){
throw new RpcException("Rpc cluster invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion()
+ " is now destroyed! Can not invoke any more.");
}
}
@Override
public String toString() {
return getInterface() + " -> " + getUrl().toString();
}
protected void checkInvokers(List<Invoker<T>> invokers, Invocation invocation) {
if (invokers == null || invokers.size() == 0) {
throw new RpcException("Failed to invoke the method "
+ invocation.getMethodName() + " in the service " + getInterface().getName()
+ ". No provider available for the service " + directory.getUrl().getServiceKey()
+ " from registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost()
+ " using the dubbo version " + Version.getVersion()
+ ". Please check if the providers have been started and registered.");
}
}
}
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailoverClusterInvoker.class);
public FailoverClusterInvoker(Directory<T> directory) {
super(directory);
}
@SuppressWarnings({ "unchecked", "rawtypes" })
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
checkInvokers(copyinvokers, invocation);
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
// retry loop.
RpcException le = null; // last exception.
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
Set<String> providers = new HashSet<String>(len);
//若是不是拋出業務異常，默認重試2次，總計調用3次，若是隻有1個服務提供者，可能會調用相同節點3次
for (int i = 0; i < len; i++) {
//重試時，進行從新選擇，避免重試時invoker列表已發生變化.
//注意：若是列表發生了變化，那麼invoked判斷會失效，由於invoker示例已經改變
if (i > 0) {
checkWheatherDestoried();
copyinvokers = list(invocation);
//從新檢查一下
checkInvokers(copyinvokers, invocation);
}
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List)invoked);
try {
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + invocation.getMethodName()
+ " in the service " + getInterface().getName()
+ " was successful by the provider " + invoker.getUrl().getAddress()
+ ", but there have been failed providers " + providers
+ " (" + providers.size() + "/" + copyinvokers.size()
+ ") from the registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost()
+ " using the dubbo version " + Version.getVersion() + ". Last error is: "
+ le.getMessage(), le);
}
return result;
} catch (RpcException e) {
if (e.isBiz()) { // biz exception.
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
+ invocation.getMethodName() + " in the service " + getInterface().getName()
+ ". Tried " + len + " times of the providers " + providers
+ " (" + providers.size() + "/" + copyinvokers.size()
+ ") from the registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
+ Version.getVersion() + ". Last error is: "
+ (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}
}
/** * 快速失敗，只發起一次調用，失敗當即報錯，一般用於非冪等性的寫操做。 * * <a href="http://en.wikipedia.org/wiki/Fail-fast">Fail-fast</a> * * @author william.liangf * @author chao.liuc */
public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T>{
public FailfastClusterInvoker(Directory<T> directory) {
super(directory);
}
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
try {
return invoker.invoke(invocation);
} catch (Throwable e) {
if (e instanceof RpcException && ((RpcException)e).isBiz()) { // biz exception.
throw (RpcException) e;
}
throw new RpcException(e instanceof RpcException ? ((RpcException)e).getCode() : 0, "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName() + " select from all providers " + invokers + " for service " + getInterface().getName() + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
}
}
}
public abstract class AbstractLoadBalance implements LoadBalance {
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.size() == 0)
return null;
if (invokers.size() == 1)
return invokers.get(0);
return doSelect(invokers, url, invocation);
}
//獲取服務提供者權重
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
protected int getWeight(Invoker<?> invoker, Invocation invocation) {
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
if (weight > 0) {
//dubbo調用方會根據服務提供方服務上線時間來調整節點權重，默認要達到10分鐘纔會分配100%權重，10分鐘之內按照時間比例來調整權重，好比服務發佈1分鐘分配的權重比例是10%,利用這種方式可讓java代碼有足夠長的熱身時間，JVM每段代碼調用超過1萬次纔會轉化爲機器指令，不然就是解釋執行。可是該段代碼有bug，在官方dubbo版本以及修復。
long timestamp = invoker.getUrl().getParameter(Constants.TIMESTAMP_KEY, 0L);
if (timestamp > 0L) {
int uptime = (int) (System.currentTimeMillis() - timestamp);
int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
if (uptime > 0 && uptime < warmup) {
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
return weight;
}
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
int ww = (int) ( (float) uptime / ( (float) warmup / (float) weight ) );
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
}
//按照每一個服務提供者節點權重隨機返回提供者
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // 總個數
int totalWeight = 0; // 總權重
boolean sameWeight = true; // 權重是否都同樣
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // 累計總權重
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false; // 計算全部權重是否同樣
}
}
if (totalWeight > 0 && ! sameWeight) {
// 若是權重不相同且權重大於0則按總權重數隨機
int offset = random.nextInt(totalWeight);
// 並肯定隨機值落在哪一個片段上
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
// 若是權重相同或權重爲0則均等隨機
return invokers.get(random.nextInt(length));
}
}
複製代碼
相關文章
- 1. Dubbo負載均衡和集羣容錯
- 2. Dubbo集羣容錯和負載均衡
- 3. 集羣下dubbo負載均衡配置
- 4. dubbo和zookeeper-負載均衡
- 5. 負載均衡集羣，Haproxy
- 6. LVS負載均衡集羣
- 7. LVS負載均衡羣集
- 8. Lvs負載均衡羣集
- 9. LVS-負載均衡集羣
- 10. 負載均衡集羣LVS
- 更多相關文章...
- • Swarm 集羣管理 - Docker教程
- • Wireshark下載安裝和使用教程 - TCP/IP教程
- • ☆技術問答集錦（13）Java Instrument原理
- • IntelliJ IDEA 代碼格式化配置和快捷鍵
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. Dubbo負載均衡和集羣容錯
- 2. Dubbo集羣容錯和負載均衡
- 3. 集羣下dubbo負載均衡配置
- 4. dubbo和zookeeper-負載均衡
- 5. 負載均衡集羣，Haproxy
- 6. LVS負載均衡集羣
- 7. LVS負載均衡羣集
- 8. Lvs負載均衡羣集
- 9. LVS-負載均衡集羣
- 10. 負載均衡集羣LVS