关于dubbo:聊聊dubbogo的failfastCluster

序

本文次要钻研一下 dubbo-go 的 failfastCluster

failfastCluster

dubbo-go-v1.4.2/cluster/cluster_impl/failfast_cluster.go

type failfastCluster struct{}

const failfast = "failfast"

func init() {extension.SetCluster(failfast, NewFailFastCluster)
}

// NewFailFastCluster ...
func NewFailFastCluster() cluster.Cluster {return &failfastCluster{}
}

func (cluster *failfastCluster) Join(directory cluster.Directory) protocol.Invoker {return newFailFastClusterInvoker(directory)
}

• failfastCluster 的 Join 办法执行 newFailFastClusterInvoker(directory)

newFailFastClusterInvoker

dubbo-go-v1.4.2/cluster/cluster_impl/failfast_cluster_invoker.go

type failfastClusterInvoker struct {baseClusterInvoker}

func newFailFastClusterInvoker(directory cluster.Directory) protocol.Invoker {
    return &failfastClusterInvoker{baseClusterInvoker: newBaseClusterInvoker(directory),
    }
}

• newFailFastClusterInvoker 办法创立了 failfastClusterInvoker

Invoke

dubbo-go-v1.4.2/cluster/cluster_impl/failfast_cluster_invoker.go

func (invoker *failfastClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {invokers := invoker.directory.List(invocation)
    err := invoker.checkInvokers(invokers, invocation)
    if err != nil {return &protocol.RPCResult{Err: err}
    }

    loadbalance := getLoadBalance(invokers[0], invocation)

    err = invoker.checkWhetherDestroyed()
    if err != nil {return &protocol.RPCResult{Err: err}
    }

    ivk := invoker.doSelect(loadbalance, invocation, invokers, nil)
    return ivk.Invoke(ctx, invocation)
}

• Invoke 办法先通过 invoker.directory.List(invocation) 获取 invokers，之后通过 getLoadBalance(invokers[0], invocation) 获取 loadbalance，再通过 invoker.doSelect(loadbalance, invocation, invokers, nil) 抉择 ivk，最初执行 ivk.Invoke(ctx, invocation)

getLoadBalance

dubbo-go-v1.4.2/cluster/cluster_impl/base_cluster_invoker.go

func getLoadBalance(invoker protocol.Invoker, invocation protocol.Invocation) cluster.LoadBalance {url := invoker.GetUrl()

    methodName := invocation.MethodName()
    //Get the service loadbalance config
    lb := url.GetParam(constant.LOADBALANCE_KEY, constant.DEFAULT_LOADBALANCE)

    //Get the service method loadbalance config if have
    if v := url.GetMethodParam(methodName, constant.LOADBALANCE_KEY, ""); len(v) > 0 {lb = v}
    return extension.GetLoadbalance(lb)
}

• getLoadBalance 办法通过 url 获取 constant.LOADBALANCE_KEY，而后通过 extension.GetLoadbalance(lb) 获取 cluster.LoadBalance

doSelect

dubbo-go-v1.4.2/cluster/cluster_impl/base_cluster_invoker.go

func (invoker *baseClusterInvoker) doSelect(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {

    var selectedInvoker protocol.Invoker
    url := invokers[0].GetUrl()
    sticky := url.GetParamBool(constant.STICKY_KEY, false)
    //Get the service method sticky config if have
    sticky = url.GetMethodParamBool(invocation.MethodName(), constant.STICKY_KEY, sticky)

    if invoker.stickyInvoker != nil && !isInvoked(invoker.stickyInvoker, invokers) {invoker.stickyInvoker = nil}

    if sticky && invoker.stickyInvoker != nil && (invoked == nil || !isInvoked(invoker.stickyInvoker, invoked)) {if invoker.availablecheck && invoker.stickyInvoker.IsAvailable() {return invoker.stickyInvoker}
    }

    selectedInvoker = invoker.doSelectInvoker(lb, invocation, invokers, invoked)

    if sticky {invoker.stickyInvoker = selectedInvoker}
    return selectedInvoker

}

• doSelect 办法先判断是否 sticky，是的话且 invoker.stickyInvoker 不为 nil 且 available，则返回 stickyInvoker，否则通过 invoker.doSelectInvoker(lb, invocation, invokers, invoked) 获取 selectedInvoker

doSelectInvoker

dubbo-go-v1.4.2/cluster/cluster_impl/base_cluster_invoker.go

func (invoker *baseClusterInvoker) doSelectInvoker(lb cluster.LoadBalance, invocation protocol.Invocation, invokers []protocol.Invoker, invoked []protocol.Invoker) protocol.Invoker {if len(invokers) == 1 {return invokers[0]
    }

    selectedInvoker := lb.Select(invokers, invocation)

    //judge to if the selectedInvoker is invoked

    if (!selectedInvoker.IsAvailable() && invoker.availablecheck) || isInvoked(selectedInvoker, invoked) {
        // do reselect
        var reslectInvokers []protocol.Invoker

        for _, invoker := range invokers {if !invoker.IsAvailable() {continue}

            if !isInvoked(invoker, invoked) {reslectInvokers = append(reslectInvokers, invoker)
            }
        }

        if len(reslectInvokers) > 0 {selectedInvoker = lb.Select(reslectInvokers, invocation)
        } else {return nil}
    }
    return selectedInvoker
}

• doSelectInvoker 办法通过 lb.Select(invokers, invocation) 抉择 selectedInvoker，如果 selectedInvoker 曾经 invoked 或者非 available，则遍历 invokers 从新结构 reslectInvokers，再通过 lb.Select(reslectInvokers, invocation) 抉择 selectedInvoker

小结

failfastCluster 的 Join 办法执行 newFailFastClusterInvoker(directory)；failfastClusterInvoker 的 Invoke 办法先通过 invoker.directory.List(invocation) 获取 invokers，之后通过 getLoadBalance(invokers[0], invocation) 获取 loadbalance，再通过 invoker.doSelect(loadbalance, invocation, invokers, nil) 抉择 ivk，最初执行 ivk.Invoke(ctx, invocation)

doc

• failfast_cluster