kubernetes源码工具包wait的使用原理剖析

阳光穿透心脏的1/2处 2022-05-16 06:07 48阅读 0赞

官方解释`JitterUntil`工具的使用说明

//管道在关闭之前 周期执行函数f
    // JitterUntil loops until stop channel is closed, running f every period.
    //
    // If jitterFactor is positive, the period is jittered before every run of f.
    // If jitterFactor is not positive, the period is unchanged and not jittered.
    //
    // If sliding is true, the period is computed after f runs. If it is false then
    // period includes the runtime for f.
    //
    // Close stopCh to stop. f may not be invoked if stop channel is already
    // closed. Pass NeverStop to if you don't want it stop.

源码目录

k8s.io/kubernetes/vendor/k8s.io/apimachinery/pkg/util/wait/wait.go

源码

func JitterUntil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) {
        var t *time.Timer
        var sawTimeout bool
    
        for {
            select {
            case <-stopCh:
                return
            default:
            }
    
            jitteredPeriod := period
            if jitterFactor > 0.0 {
                jitteredPeriod = Jitter(period, jitterFactor)
            }
    
            if !sliding {
                t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
            }
    
            func() {
                defer runtime.HandleCrash()
                f()
            }()
    
            if sliding {
                t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout)
            }
    
            // NOTE: b/c there is no priority selection in golang
            // it is possible for this to race, meaning we could
            // trigger t.C and stopCh, and t.C select falls through.
            // In order to mitigate we re-check stopCh at the beginning
            // of every loop to prevent extra executions of f().
            select {
            case <-stopCh:
                return
            case <-t.C:
                sawTimeout = true
            }
        }
    }

// resetOrReuseTimer avoids allocating a new timer if one is already in use. // Not safe for multiple threads.
    func resetOrReuseTimer(t *time.Timer, d time.Duration, sawTimeout bool) *time.Timer {
        if t == nil {
            return time.NewTimer(d)
        }
        if !t.Stop() && !sawTimeout {
            <-t.C
        }
        t.Reset(d)
        return t
    }

`sliding` 滑动处理 如果为`true`,计时包括执行 f 函数的花费时间，如果为`false`,不包含执行时间

--------------------

简单使用例子

func main() {
    
        stop := make(chan struct{})
    
        go func() {
            fmt.Printf("sleep 10!\n")
    
            time.Sleep(10 * time.Second)
            var a struct{}
            stop <- a
        }()
    
        f := func() {
            time.Sleep(1 * time.Second)
            fmt.Printf("hello world!\n")
        }
        //100000000
        Until(f, 1*time.Second, stop)
    
        //NonSlidingUntil(f, 1*time.Second, stop)
    
        fmt.Println("the chan closed")
    
    }

执行结果

➜  go run jitter.go
    sleep 10!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    the chan closed

--------------------

func main() {
    
        stop := make(chan struct{})
    
        go func() {
            fmt.Printf("sleep 10!\n")
    
            time.Sleep(10 * time.Second)
            var a struct{}
            stop <- a
        }()
    
        f := func() {
            time.Sleep(1 * time.Second)
            fmt.Printf("hello world!\n")
        }
        //100000000
        //Until(f, 1*time.Second, stop)
    
        NonSlidingUntil(f, 1*time.Second, stop)
    
        fmt.Println("the chan closed")
    
    }

执行结果

➜  go run jitter.go
    sleep 10!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    hello world!
    the chan closed

kubernetes 的源码好多地方都用到这个函数