关于dubbo:聊聊dubbogo的failbackCluster

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

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

type failbackCluster struct{}

const failback = "failback"

func init() {extension.SetCluster(failback, NewFailbackCluster)
}

// NewFailbackCluster ...
func NewFailbackCluster() cluster.Cluster {return &failbackCluster{}
}

func (cluster *failbackCluster) Join(directory cluster.Directory) protocol.Invoker {return newFailbackClusterInvoker(directory)
}

failbackCluster 的 join 办法执行 newFailbackClusterInvoker

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

type failbackClusterInvoker struct {
    baseClusterInvoker

    once          sync.Once
    ticker        *time.Ticker
    maxRetries    int64
    failbackTasks int64
    taskList      *queue.Queue
}

func newFailbackClusterInvoker(directory cluster.Directory) protocol.Invoker {
    invoker := &failbackClusterInvoker{baseClusterInvoker: newBaseClusterInvoker(directory),
    }
    retriesConfig := invoker.GetUrl().GetParam(constant.RETRIES_KEY, constant.DEFAULT_FAILBACK_TIMES)
    retries, err := strconv.Atoi(retriesConfig)
    if err != nil || retries < 0 {logger.Error("Your retries config is invalid,pls do a check. And will use the default fail back times configuration instead.")
        retries = constant.DEFAULT_FAILBACK_TIMES_INT
    }

    failbackTasksConfig := invoker.GetUrl().GetParamInt(constant.FAIL_BACK_TASKS_KEY, constant.DEFAULT_FAILBACK_TASKS)
    if failbackTasksConfig <= 0 {failbackTasksConfig = constant.DEFAULT_FAILBACK_TASKS}
    invoker.maxRetries = int64(retries)
    invoker.failbackTasks = failbackTasksConfig
    return invoker
}

newFailbackClusterInvoker 办法创立 failbackClusterInvoker，并设置其 maxRetries、failbackTasks 属性

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

func (invoker *failbackClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {invokers := invoker.directory.List(invocation)
    err := invoker.checkInvokers(invokers, invocation)
    if err != nil {
        logger.Errorf("Failed to invoke the method %v in the service %v, wait for retry in background. Ignored exception: %v.\n",
            invocation.MethodName(), invoker.GetUrl().Service(), err)
        return &protocol.RPCResult{}}
    url := invokers[0].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, ""); v !="" {lb = v}
    loadbalance := extension.GetLoadbalance(lb)

    invoked := make([]protocol.Invoker, 0, len(invokers))
    var result protocol.Result

    ivk := invoker.doSelect(loadbalance, invocation, invokers, invoked)
    //DO INVOKE
    result = ivk.Invoke(ctx, invocation)
    if result.Error() != nil {invoker.once.Do(func() {invoker.taskList = queue.New(invoker.failbackTasks)
            go invoker.process(ctx)
        })

        taskLen := invoker.taskList.Len()
        if taskLen >= invoker.failbackTasks {logger.Warnf("tasklist is too full > %d.\n", taskLen)
            return &protocol.RPCResult{}}

        timerTask := newRetryTimerTask(loadbalance, invocation, invokers, ivk)
        invoker.taskList.Put(timerTask)

        logger.Errorf("Failback to invoke the method %v in the service %v, wait for retry in background. Ignored exception: %v.\n",
            methodName, url.Service(), result.Error().Error())
        // ignore
        return &protocol.RPCResult{}}

    return result
}

Invoke 办法先通过 invoker.directory.List(invocation) 获取 invokers，之后通过 extension.GetLoadbalance(lb) 获取 loadbalance，而后通过 invoker.doSelect(loadbalance, invocation, invokers, invoked) 抉择 invoker，之后执行其 Invoke 办法，如果出现异常则设置 invoker.taskList，异步执行 invoker.process(ctx)，之后通过 newRetryTimerTask 创立 timerTask，增加到 invoker.taskList

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

func (invoker *failbackClusterInvoker) Destroy() {invoker.baseClusterInvoker.Destroy()

    // stop ticker
    if invoker.ticker != nil {invoker.ticker.Stop()
    }

    _ = invoker.taskList.Dispose()}

Destroy 办法执行 invoker.baseClusterInvoker.Destroy()、invoker.ticker.Stop()、invoker.taskList.Dispose()

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

func (invoker *failbackClusterInvoker) process(ctx context.Context) {invoker.ticker = time.NewTicker(time.Second * 1)
    for range invoker.ticker.C {
        // check each timeout task and re-run
        for {value, err := invoker.taskList.Peek()
            if err == queue.ErrDisposed {return}
            if err == queue.ErrEmptyQueue {break}

            retryTask := value.(*retryTimerTask)
            if time.Since(retryTask.lastT).Seconds() < 5 {break}

            // ignore return. the get must success.
            _, err = invoker.taskList.Get(1)
            if err != nil {logger.Warnf("get task found err: %v\n", err)
                break
            }

            go func(retryTask *retryTimerTask) {invoked := make([]protocol.Invoker, 0)
                invoked = append(invoked, retryTask.lastInvoker)

                retryInvoker := invoker.doSelect(retryTask.loadbalance, retryTask.invocation, retryTask.invokers, invoked)
                var result protocol.Result
                result = retryInvoker.Invoke(ctx, retryTask.invocation)
                if result.Error() != nil {
                    retryTask.lastInvoker = retryInvoker
                    invoker.checkRetry(retryTask, result.Error())
                }
            }(retryTask)

        }
    }
}

process 办法通过 time.NewTicker(time.Second * 1) 创立 invoker.ticker，之后从 invoker.taskList.Peek() 获取 retryTask( 之后 Get 办法进行 poll)，而后异步执行 invoker.doSelect(retryTask.loadbalance, retryTask.invocation, retryTask.invokers, invoked) 选取 retryInvoker，而后执行 retryInvoker.Invoke(ctx, retryTask.invocation)；如果执行出现异常，则通过 invoker.checkRetry(retryTask, result.Error()) 进行 check

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

func (invoker *failbackClusterInvoker) checkRetry(retryTask *retryTimerTask, err error) {
    logger.Errorf("Failed retry to invoke the method %v in the service %v, wait again. The exception: %v.\n",
        retryTask.invocation.MethodName(), invoker.GetUrl().Service(), err.Error())
    retryTask.retries++
    retryTask.lastT = time.Now()
    if retryTask.retries > invoker.maxRetries {logger.Errorf("Failed retry times exceed threshold (%v), We have to abandon, invocation-> %v.\n",
            retryTask.retries, retryTask.invocation)
    } else {invoker.taskList.Put(retryTask)
    }
}

checkRetry 办法会递增 retryTask.retries，而后判断是否超过 invoker.maxRetries，超过则记录 error 日志，不超过则再次将 retryTask 增加到 invoker.taskList

newFailbackClusterInvoker 办法创立 failbackClusterInvoker，并设置其 maxRetries、failbackTasks 属性；其 Invoke 办法先通过 invoker.directory.List(invocation) 获取 invokers，之后通过 extension.GetLoadbalance(lb) 获取 loadbalance，而后通过 invoker.doSelect(loadbalance, invocation, invokers, invoked) 抉择 invoker，之后执行其 Invoke 办法，如果出现异常则设置 invoker.taskList，异步执行 invoker.process(ctx)，之后通过 newRetryTimerTask 创立 timerTask，增加到 invoker.taskList

failbackCluster 疏忽 result，针对失败的会退出队列重试 maxRetries 次，适宜 fireAndForget 的通信模式

failback_cluster

关于dubbo:聊聊dubbogo的failbackCluster

序

failbackCluster

newFailbackClusterInvoker

Invoke

Destroy

process

checkRetry

小结

doc