您的位置 首页 golang

client-go和golang源码中的技巧

智云科技发布于 2022年7月27日评论关闭阅读(299)

client-go中有很多比较有意思的实现，如定时器，同步机制等，可以作为移植使用。下面就遇到的一些技术讲解，首先看第一个：

  • sets.String (k8s.io/apimachinery/pkg/util/sets/string.go)

实现了对golang map的key的处理，如计算交集，并集等。实际中可能会遇到需要判断两个map的key是否重合的场景，此时可以使用下述方式实现，sets.StringKeySet函数将入参的map的key抽取成一个String类型，这样就可以使用String的方法操作key

ps:更多功能参见源码 

 package main

 import  (
    "fmt"
    "k8s.io/apimachinery/pkg/util/sets"
)

func main(){
    map1 := map[string]int{"aaa":1,"bbb":2,"ccc":3}
    map2 := map[string]int{"ccc":1,"ddd":2,"eee":3}
    newmap1 := sets.StringKeySet(map1)
    newmap2 := sets.StringKeySet(map2)
    fmt.Println(newmap1.List(),newmap2.List())
    fmt.Println(newmap1.HasAny(newmap2.List()...)) //3个点用于把数组打散为单个元素
}

结果：true
  • 同步机制sync.Mutex(golang 内置方法)，用于数据同步

有2个方法： 

 func (m *Mutex)  Lock ()
func (m *Mutex) Unlock()

类似C语言线程的互斥锁，用于对数据进行加解锁操作。当数据被加锁后，未获得该锁的程序将无法读取被加锁的数据。从下面例子可以看出在数据被解锁前其他协程无法对该数据进行读写操作。

ps: read data 的数据也可能为“ data ” 

 package main

import (
    "fmt"
    "sync"
)

type LockTest struct {
    l sync.Mutex
    data string
}

func main(){
    lockTest := LockTest{sync.Mutex{},"data"}
    go func() {
        lockTest.l.Lock()
        fmt.Println("sleep begin")
        time.Sleep(time.Second*2)
        fmt.Println("sleep end")
        lockTest.l.Unlock()
    }()
    
    time.Sleep(time.Second*1)
    
    go func() {
        lockTest.l.Lock()
        fmt.Println("read data:",lockTest.data)
        lockTest.l.Unlock()
    }()

    go func() {
        lockTest.l.Lock()
        fmt.Println("write data begin")
        lockTest.data="new data"
        fmt.Println("write data end")
        lockTest.l.Unlock()
    }()

    time.Sleep(time.Second*5)
}

结果：
sleep begin
sleep end
write data begin
write data end
read data: new data
  • sync.RWMutex(golang 内置方法)，用于数据同步

读写锁，含4个方法，前2个为读锁，后2个为写锁，使用时要一一对应。写锁会阻塞读写操作，读锁不会阻塞写操作，读锁可以有多个，读锁之间不会相互阻塞，适用于读多写少的场景。因此如果单纯使用RWMutex.Lock/RWMutex.UnLock与使用Mutex.Lock/Mutex.UnLock效果相同 

 func (rw *RWMutex) RLock()
func (rw *RWMutex) RUnlock()
func (rw *RWMutex) Lock()
func (rw *RWMutex) Unlock()

读写锁一般是读锁和写锁结合使用的。在有写锁的时候，读锁会被阻塞，等待写锁释放后才能进行读操作。

ps:sync.Mutex和sync.RWMutex一般都是内置在结构体中使用，用于保护本结构体的数据 

 package main

import (
    "fmt"
    "sync"
)
type LockTest struct {
    l sync.RWMutex
    data string
}

func main(){
    lockTest := LockTest{sync.RWMutex{},"data"}
    go func() {
        lockTest.l.Lock()
        fmt.Println("write data begin")
        lockTest.data="new data"
        time.Sleep(time.Second*3)
        fmt.Println("write data end")
        lockTest.l.Unlock()
    }()

    time.Sleep(time.Second*1)

    go func() {
        lockTest.l.RLock()  //阻塞等待写锁释放
        fmt.Println("read begin")
        fmt.Println("read data:",lockTest.data)
        fmt.Println("read begin")
        lockTest.l.RUnlock()
    }()

    time.Sleep(time.Second*5)
}

结果：
write data begin
write data end
read begin
read data: new data
read begin
  • sync.Cond(golang 内置方法)，用于条件变量

sync.Cond用于条件等待，在满足某些条件时程序才能继续执行。它包含如下3个方法：Wait()会挂起其所在的协程等待 Signal ()或Broadcast()的唤醒。 

 func (c *Cond) Wait() 
func (c *Cond) Signal()
func (c *Cond) Broadcast()

官方推荐的典型用法如下。由于唤醒协程并不意味着条件已就绪，因此在唤醒后需要检测是否本协程的条件已经满足。 

 c.L.Lock()
for !condition() {
    c.Wait()
}
... make use of condition ...
c.L.Unlock()

使用Signal()唤醒的方式如下，Signal()用于当次唤醒一个协程。如果注释掉下例中的Signal()，那么两个协程会一直Wait()，并不会继续执行。 

 package main

import (
    "fmt"
    "sync"
)

func main(){
    l := sync.Mutex{}
    c := sync.NewCond(&l)
    condition1 := false
    condition2 := false
    go func() {
        c.L.Lock()
        for !condition1 {
            c.Wait()
        }
        fmt.Println("condition1=true,run1")
        c.L.Unlock()
    }()

    go func() {
        c.L.Lock()
        for !condition2 {
            c.Wait()
        }
        fmt.Println("condition2=true,run2")
        c.L.Unlock()
    }()

    time.Sleep(time.Second*1)
    fmt.Println("signal-1")
    condition1=true
    c.Signal()
    time.Sleep(time.Second*1)
    fmt.Println("signal-2")
    condition2=true
    c.Signal()
    time.Sleep(time.Second*10)
}


结果：
signal-1
condition1=true,run1
signal-2
condition2=true,run2

使用Signal()唤醒协程时需要注意，在多个协程等待时，该函数并没有指定需要唤醒哪一个协程。下面程序的输出可能为“ condition1=true,run1 ”也可能为“ condition2=true,run2 ”。因此Signal一般适用于仅有一个协程等待的情况，否则可能造成混乱。 

 package main

import (
    "fmt"
    "sync"
)

func main(){
    l := sync.Mutex{}
    c := sync.NewCond(&l)
    condition1 := false
    condition2 := false
    go func() {
        c.L.Lock()
        for !condition1 {
            c.Wait()
        }
        fmt.Println("condition1=true,run1")
        c.L.Unlock()
    }()

    go func() {
        c.L.Lock()
        for !condition2 {
            c.Wait()
        }
        fmt.Println("condition2=true,run2")
        c.L.Unlock()
    }()
    time.Sleep(time.Second*1)
    condition1=true
    condition2=true
    c.Signal()
    time.Sleep(time.Second*10)
}

Broadcast()比较简单，即唤醒所有等待的协程 

 package main

import (
    "fmt"
    "sync"
)

func main(){
    l := sync.Mutex{}
    c := sync.NewCond(&l)
    condition1 := false
    condition2 := false
    go func() {
        c.L.Lock()
        for !condition1 {
            c.Wait()
        }
        fmt.Println("condition1=true,run1")
        c.L.Unlock()
    }()

    go func() {
        c.L.Lock()
        for !condition2 {
            c.Wait()
        }
        fmt.Println("condition2=true,run2")
        c.L.Unlock()
    }()
    time.Sleep(time.Second*1)
    condition1=true
    condition2=true
    c.Broadcast()
    time.Sleep(time.Second*10)
}

结果：
condition1=true,run1
condition2=true,run2
  • sync.waitgroup，用于等待协程执行完成

sync.waitgroup有如下3个方法，Add(delta int)入参表示需要等待的协程的个数，如2表示需要等待2个协程完成；Done()表示该协程结束；Wait()用于阻塞主协程，等待所有协程结束后释放。 

 func (wg *WaitGroup) Add(delta int)
func (wg *WaitGroup) Done()
func (wg *WaitGroup) Wait()

举例如下，启动10个协程，Wait()会阻塞，直到所有的协程执行Done()。

ps: Add(delta int)函数的入参很重要，入参大于实际需要等待的协程会导致主协程一致阻塞，小于需要等待的协程会导致某些协程提前退出 

 import (
    "fmt"
    "sync"
)

func main(){
    wg := sync.WaitGroup{}
    wg.Add(10)

    for i := 0; i < 10; i++ {
        go func(i int) {
            defer wg.Done()
            fmt.Print(i, " ")
        }(i)
    }

    wg.Wait()
}

结果：
9 4 0 1 2 3 6 5 7 8
  • 协程间使用chan进行同步

下例中使用chan实现主协程控制write，并使用write控制read。协程关闭使用close()函数

ps：使用chan进行协程同步一般将chan作为入参传入，或在函数内部实现协程间的同步。为方便验证，下面例子将所有chan作为全局变量 

 package main

import (
    "fmt"
    "sync"
)
var speakCh = make(chan string)
var stopReadChan = make(chan struct{})
var stopWriteChan = make(chan struct{})

func readChan(stopCh <-chan struct{}){
    for {
        select {
        case words := <- speakCh:
            fmt.Println("received:",words)
        case <- stopCh:
            fmt.Println("stop read!")
            return
        }
    }
}

func writeChan(stopCh <-chan struct{}){
    for {
        select {
        case <- stopCh:
            fmt.Println("stop write!")
            close(stopReadChan）
            return
        default:
        }
        speakCh <- "hi"
        time.Sleep(time.Second*2)
    }
}

func main(){
    go readChan(stopReadChan)
    go writeChan(stopWriteChan)

    time.Sleep(time.Second*6)
    close(stopWriteChan)
    time.Sleep(time.Second*6)
}

结果：
received: hi
received: hi
received: hi
stop write!
stop read!
  • 协程间使用context进行同步

context用于对协程进行管理，如主动退出协程，超时退出协程等，可以看作是使用chan管理协程的扩展。在使用时首先创建一个context，使用cancel()可以取消context，并使用Done()返回的chan管理协程。

官方推荐的用法如下： 

 func Stream(ctx context.Context, out chan<- Value) error {
    for {
        v, err := DoSomething(ctx)
        if err != nil {
            return err
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case out <- v:
        }
    }
}

下例中使用context.WithCancel创建一个context，使用cancel()给这一组context发送信号，在协程中使用Done()处理退出事件。 

 package main

import (
    "fmt"
    "context"
)

func main(){
    ctx,cancel := context.WithCancel(context.Background())
    go testCtx(ctx,"ctx1")
    go testCtx(ctx,"ctx2")
    go testCtx(ctx,"ctx3")
    time.Sleep(time.Second*3)
    cancel()

    time.Sleep(time.Second*5)
}

func testCtx(ctx context.Context, name string) error{
    for {
        select {
        case <-ctx.Done():
            fmt.Println("ctx.Done:",name)
            return ctx.err()
        default:
            fmt.Println("default:",name)
            time.Sleep(time.Second*2)
        }
    }
}

结果：
default: ctx1
default: ctx3
default: ctx2
default: ctx3
default: ctx1
default: ctx2
ctx.Done: ctx1
ctx.Done: ctx3
ctx.Done: ctx2

创建context的方式如下，其余三个可以看作是WithCancel的扩展 

 func WithCancel(parent Context) (ctx Context, cancel CancelFunc)              //需要主动取消context
func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)   //在deadline时间点后取消context
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) //在超时后取消context
func WithValue(parent Context, key, val interface{}) Context

再看一个WithTimeout的例子，下面设置context的超时时间为3s且没有主动cancel()，3s超时后可以看到该context对应的协程正常退出 

 func main(){
    ctx,_ := context.WithTimeout(context.Background(),time.Second*3)
    go testCtx(ctx,"ctx1")
    go testCtx(ctx,"ctx2")
    go testCtx(ctx,"ctx3")
    time.Sleep(time.Second*5)
}

结果：
default: ctx3
default: ctx1
default: ctx2
default: ctx3
default: ctx1
default: ctx2
ctx.Done: ctx3
ctx.Done: ctx2
ctx.Done: ctx1

context可以看作是一个树，当cancel一个context时，会同时cancle它的子context。下面首先创建一个ctx，然后在此ctx下面创建一个subctx。当执行cancle() ctx时会同时cancel() 该的subctx。

context.Background()就是已经实现的首个context。 

 func main(){
    ctx,cancel := context.WithCancel(context.Background())
    subctx,_ := context.WithCancel(ctx)
    go testCtx(ctx,"ctx1")
    go testCtx(subctx,"subctx1")
    go testCtx(subctx,"subctx2")
    time.Sleep(time.Second*3)
    canclel()

    time.Sleep(time.Second*10)
}

结果：
default: subctx2
default: ctx1
default: subctx1
default: subctx2
default: ctx1
default: subctx1
timeout
ctx.Done: ctx1
ctx.Done: subctx1
ctx.Done: subctx2

下例中仅cancel() subctx，可以看到并没有影响subctx的parent。 

 func main(){
    ctx, _:= context.WithCancel(context.Background())
    subctx,subcancel := context.WithCancel(ctx)
    go testCtx(ctx,"ctx1")
    go testCtx(subctx,"subctx1")
    go testCtx(subctx,"subctx2")
    time.Sleep(time.Second*3)
    subcancel()

    time.Sleep(time.Second*10)
}

结果：
default: subctx1
default: subctx2
default: ctx1
default: ctx1
default: subctx1
default: subctx2
timeout
ctx.Done: subctx2
default: ctx1
ctx.Done: subctx1
default: ctx1
default: ctx1
default: ctx1
default: ctx1
  • wait.Group (k8s.io/apimachinery/pkg/util/wait/wait.go)

client-go中的wait.Group创造性地将sync.WaitGroup与chan和ctx结合，实现了协程间同步和等待全部Group中的协程结束的功能。由于StartWithChannel和StartWithContext的入参函数类型比较固定，因此使用上并不通用，但可以作为参考，在实际中扩展使用。下例中给出了简单用法。 

 func (g *Group) Wait() 
func (g *Group) StartWithChannel(stopCh <-chan struct{}, f func(stopCh <-chan struct{}))
func (g *Group) StartWithContext(ctx context.Context, f func(context.Context))  
 func main(){
    f1:= func(ctx context.Context) {
        for {
            select {
            case <- ctx.Done():
                return
            default:
                fmt.Println("hi11")
                time.Sleep(time.Second)
            }
        }
    }
    wg := wait.Group{}
    ctx, cancel := context.WithCancel(context.Background())
    wg.StartWithContext(ctx,f1)
    time.Sleep(time.Second*3)
    cancel()
    wg.Wait()
}

结果：
hi
hi
hi

  • 定时器ticker定时器

首先看一下一般使用的定时器，client-go中比较复杂的定时器也是在此基础上封装的。下面例子中给出的是ticker定时器，它会按照一定的时间频率往Ticker.C中发time.Time类型的数据，可以在协程中通过判断Ticker.C来执行定时任务。下例来自官方，实现每秒执行一次打印， 

 import (
    "fmt"
    "time"
)

func main(){
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()
    done := make(chan bool)
    go func() {
        time.Sleep(10 * time.Second)
        done <- true
    }()
    for {
        select {
        case <-done:
            fmt.Println("Done!")
            return
        case t := <-ticker.C:
            fmt.Println("Current time: ", t)
        }
    }
}

结果：
Current time:  2019-07-04 14:30:37.9088968 +0800 CST m=+5.328291301
Current time:  2019-07-04 14:30:38.9089349 +0800 CST m=+6.328328801
Current time:  2019-07-04 14:30:39.9101415 +0800 CST m=+7.329534901
Current time:  2019-07-04 14:30:40.9095174 +0800 CST m=+8.328910201
Current time:  2019-07-04 14:30:41.9092961 +0800 CST m=+9.328688301
Current time:  2019-07-04 14:30:42.9087682 +0800 CST m=+10.328159801
Current time:  2019-07-04 14:30:43.9088604 +0800 CST m=+11.328251401
Current time:  2019-07-04 14:30:44.909609 +0800 CST m=+12.328999501
Current time:  2019-07-04 14:30:45.9094782 +0800 CST m=+13.328868101
Current time:  2019-07-04 14:30:46.909006 +0800 CST m=+14.328395401
Done!

需要注意的是使用ticker并不能保证程序被精确性调度，如果程序的执行时间大于ticker的调度周期，那么程序的触发周期会发生偏差(可能由于系统cpu占用过高，网络延迟等原因)。如下面例子中，ticker触发周期为1s，但程序执行大于2s，此时会出现程序执行频率不一致的情况。适用于周期性触发一个任务。 

 func main(){
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()
    done := make(chan bool)
    go func() {
        time.Sleep(10 * time.Second)
        done <- true
    }()
    for {
        select {
        case <-done:
            fmt.Println("Done!")
            return
        case t := <-ticker.C:
            time.Sleep(time.Second*2)
            fmt.Println("Current time: ", t)
        }
    }
}

结果：
Current time:  2019-07-04 14:56:52.5446526 +0800 CST m=+5.281916601  
Current time:  2019-07-04 14:56:53.5452488 +0800 CST m=+6.282512201  //和上一条相差1s，但和下一条相差2s
Current time:  2019-07-04 14:56:55.5443528 +0800 CST m=+8.281615101
Current time:  2019-07-04 14:56:57.5449183 +0800 CST m=+10.282179401
Current time:  2019-07-04 14:56:59.5448671 +0800 CST m=+12.282127101
Done!
  • timer定时器

timer的机制和ticker相同，在定时器超时后往一个chan中发送time.Time数据。不同的是ticker可以周期性调度，timer只会执行一次，如果需要重复调度，需要使用Reset函数重置timer。利用该机制，可以在同一个timer上以不同间隔调度程序。 

 func main(){
    timer := time.NewTimer(time.Second)
    defer timer.Stop()
    t := <-timer.C
    fmt.Println("Current time: ", t)
    timer.Reset(time.Second*2)
    t = <-timer.C
    fmt.Println("Current time: ", t)
    timer.Reset(time.Second*3)
    t = <-timer.C
    fmt.Println("Current time: ", t)
}

结果：
Current time:  2019-07-04 15:47:01.7518201 +0800 CST m=+5.312710501
Current time:  2019-07-04 15:47:03.7766692 +0800 CST m=+7.337558501
Current time:  2019-07-04 15:47:06.7770913 +0800 CST m=+10.337978901

使用timer需要注意Reset函数只能在timer超时后使用，否则将无效。因为Timer.C的长度只有1，如果前面一个定时器结束前执行了Reset，那么前面的定时器会被取消。具体可以参见这里 

 func NewTimer(d Duration) *Timer {
    c := make(chan Time, 1)
    ...
}

下面例子中可以看出，多次执行Reset并不会多次触发定时任务，在前一个定时器超时前执行Reset，会取消前一个定时器并以Reset中的duration开始计时。 

 func main(){
    fmt.Println("now time: "time.Now())
    timer := time.NewTimer(time.Second*5)
    
    defer timer.Stop()
    timer.Reset(time.Second*2)
    timer.Reset(time.Second*2)
    timer.Reset(time.Second*2)


    go func() {
        for ; ;  {
            select {
            case t:=<- timer.C:
                fmt.Println("Current time: ", t)
            }
        }
    }()
    
    time.Sleep(time.Second*10)
}

结果：
now time:      2019-07-04 16:16:31.7246084 +0800 CST m=+4.281414201
Current time:  2019-07-04 16:16:33.7505395 +0800 CST m=+6.307344201

官方推荐的用法如下，由于没有加锁，此方法不能在多个协程中同时使用。 

 if !t.Stop() {
    <-t.C
}
t.Reset(d)

func AfterFunc(d Duration, f func()) *Timer函数用于在d时间超时后，执行f函数。注意返回的timer需要手动stop 

 timer := time.AfterFunc(time.Second*5, func() {
    fmt.Println("timeout")
})

time.Sleep(time.Second*6)
timer.Stop()

更多timer的用法可以参见官方文档

  • wait实现 (k8s.io/apimachinery/pkg/util/wait/wait.go)
    • wait中实现了很多与定时相关的函数，首先来看第一组： 
 func Forever(f func(), period time.Duration) 
func Until(f func(), period time.Duration, stopCh <-chan struct{}) 
func UntilWithContext(ctx context.Context, f func(context.Context), period time.Duration) 
func NonSlidingUntil(f func(), period time.Duration, stopCh <-chan struct{}) 
func NonSlidingUntilWithContext(ctx context.Context, f func(context.Context), period time.Duration)

Until函数每period会调度f函数，如果stopCh中有停止信号，则退出。当程序运行时间超过period时，也不会退出调度循环，该特性和Ticker相同。底层使用Timer实现。

Until和NonSlidingUntil为一对，UntilWithContext和NonSlidingUntilWithContext为一对，区别只是定时器启动时间点不同，可以简单用下图表示：

这两种(带“NonSliding”前缀的)函数在处理正常程序时没有什么区别，但在一些场景下会有不同的地方。下面例子中使用wait.NonSlidingUntil处理的程序中sleep了2s，这可以表示程序因为某种原因导致超出正常处理时间。此时可以看到结果中的“num 1”和“num 2”是同时调用的 

 func main(){
    first := true
    num := 0
    stopCh:=make(chan struct{} )
    
    go func() {
        time.Sleep(time.Second*10)
        close(stopCh)
        fmt.Println("done")
    }()

    go wait.NonSlidingUntil(func(){
        if true == first{
            time.Sleep(time.Second*2)
            first=false
        }
        num = num + 1
        fmt.Println("num:",num,"time",time.Now())
    },time.Second*1,stopCh)

    time.Sleep(time.Second*100)
}

结果：
num: 1 time 2019-07-04 21:05:59.5298524 +0800 CST m=+26.277103101
num: 2 time 2019-07-04 21:05:59.554999 +0800 CST m=+26.302249701
num: 3 time 2019-07-04 21:06:00.5559679 +0800 CST m=+27.303218601
num: 4 time 2019-07-04 21:06:01.5566608 +0800 CST m=+28.303911501

将上述程序的wait.NonSlidingUntil替换为wait.Until，得到如下结果，可以看到首次(异常)和第二次(正常)的间隔正好是wait.Until中设置的调度周期，即1s。

ps：大部分场景下两者使用上并没有什么不同，毕竟正常情况下程序运行时间必然小于程序调度周期。如果需要在程序处理延时的情况下尽快进行下一次调度，则选择带”NonSliding“前缀的函数 

 结果：
num: 1 time 2019-07-04 21:09:14.9643889 +0800 CST m=+2.010865201
num: 2 time 2019-07-04 21:09:15.9935285 +0800 CST m=+3.040004801
num: 3 time 2019-07-04 21:09:16.9956846 +0800 CST m=+4.042160901
  • func Forever(f func(), period time.Duration)

该函数比较简单，就是取消了用于控制Until停止的stopCh。以永远不停止的方式周期性执行f函数

  • func ExponentialBackoff(backoff Backoff, condition ConditionFunc) error

ExponentialBackoff可以实现在函数执行错误后实现以指数退避方式的延时重试。ExponentialBackoff内部使用的是time.Sleep

ExponentialBackoff的首个入参Backoff如下：

  • Duration：表示初始的延时时间
  • Factor：指数退避的因子
  • Jitter：可以看作是偏差因子，该值越大，每次重试的延时的可选区间越大
  • Steps：指数退避的步数，可以看作程序的最大重试次数
  • Cap：用于在Factor非0时限制最大延时时间和最大重试次数，为0表示不限制最大延时时间 
 type Backoff struct {
    // The  initial  duration.
    Duration time.Duration
    // Duration is multiplied by factor each iteration. Must be greater
    // than or equal to zero.
    Factor float64
    // The amount of jitter applied each iteration. Jitter is applied after
    // cap.
    Jitter float64
    // The number of steps before duration stops changing. If zero, initial
    // duration is always used. Used for exponential backoff in combination
    // with Factor.
    Steps int
    // The returned duration will never be greater than cap *before* jitter
    // is applied. The actual maximum cap is `cap * (1.0 + jitter)`.
    Cap time.Duration
}

第二个参数ConditionFunc表示运行的函数，返回的bool值表示该函数是否执行成功，如果执行成功则会退出指数退避 

 type ConditionFunc func() (done bool, err error)

下面做几组测试：

=> 当Factor和Jitter都为0时，可以看到调度周期是相同的，即Duration的值(1s)。 

 import (
    "fmt"
    "k8s.io/apimachinery/pkg/util/wait"
    "time"
)


func main(){
    var DefaultRetry = wait.Backoff{
        Steps:    5,
        Duration: 1 * time.Second,
        Factor:   0,
        Jitter:   0,
    }

    fmt.Println(wait.ExponentialBackoff(DefaultRetry,func() (bool, error){
        fmt.Println(time.Now())
        return false,nil
    }))
}

结果：
2019-07-05 10:17:33.9610108 +0800 CST m=+0.079831101
2019-07-05 10:17:34.961132 +0800 CST m=+1.079952301
2019-07-05 10:17:35.961512 +0800 CST m=+2.080332301
2019-07-05 10:17:36.9625144 +0800 CST m=+3.081334701
2019-07-05 10:17:37.9636334 +0800 CST m=+4.082453701
timed out waiting for the condition

=> 先看Jitter对duration的影响，Jitter(duration, b.Jitter)的计算方式如下，如果入参的Factor为0，而Jitter非0，则将Factor调整为1。rand.Float64()为[0.0,1.0)的伪随机数。

将Jitter调整为0.5，根据下面计算方式预期duration为[1s,1.5s)。运行程序得出如下结果，观察可以发现，duration大概是1.4s 

 if maxFactor <= 0.0 {
    maxFactor = 1.0
}
wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration))  
 var DefaultRetry = wait.Backoff{
    Steps:    5,
    Duration: 1 * time.Second,
    Factor:   0,
    Jitter:   0.5,
}

结果：
2019-07-05 10:21:49.5993445 +0800 CST m=+2.382669101
2019-07-05 10:21:50.9026701 +0800 CST m=+3.685994701
2019-07-05 10:21:52.3759019 +0800 CST m=+5.159226401
2019-07-05 10:21:53.7086265 +0800 CST m=+6.491951001
2019-07-05 10:21:54.9283913 +0800 CST m=+7.711715901
timed out waiting for the condition

=> Factor非0且Jitter为0时，对duration的调整如下 

 if b.Factor != 0 {
    b.Duration = time.Duration(float64(b.Duration) * b.Factor)
    if b.Cap > 0 && b.Duration > b.Cap {
        b.Duration = b.Cap
        b.Steps = 0
    }
}

从公式中可以得出，Factor对程序执行的延的影响如下，可以看到Factor为1时并没有什么作用 

 duration(1) = duration
duration(2) = Factor * duration(1)
duration(3) = Factor * duration(2)
...
duration(n) = Factor * duration(n-1)

Factor为1时，可以看到函数执行间隔均为1s 

 var DefaultRetry = wait.Backoff{
    Steps:    5,
    Duration: 1 * time.Second,
    Factor:   1,
    Jitter:   0,
}

结果：
2019-07-05 10:28:50.8481017 +0800 CST m=+2.363983901
2019-07-05 10:28:51.8482274 +0800 CST m=+3.364109601
2019-07-05 10:28:52.8482359 +0800 CST m=+4.364118201
2019-07-05 10:28:53.848687 +0800 CST m=+5.364569301
2019-07-05 10:28:54.849409 +0800 CST m=+6.365291201
timed out waiting for the condition

调整Factor为3，预期延时时间为1s，3s，9s，27s，从测试结果看与预期相符 

 var DefaultRetry = wait.Backoff{
    Steps:    5,
    Duration: 1 * time.Second,
    Factor:   3,
    Jitter:   0,
}

结果：
2019-07-05 10:35:06.9030165 +0800 CST m=+0.077746101
2019-07-05 10:35:07.9038392 +0800 CST m=+1.078568701
2019-07-05 10:35:10.9038733 +0800 CST m=+4.078602901
2019-07-05 10:35:19.9042141 +0800 CST m=+13.078943601
2019-07-05 10:35:46.904647 +0800 CST m=+40.079376501
timed out waiting for the condition

=> 当Factor和Jitter非0时的延迟计算方式如下： 

     save_duration(0) = duration
duration(1) =  Jitter(save_duration(0) , b.Jitter)
    save_duration(1) = Factor * save_duration(0) 

duration(2) = Jitter(save_duration(1), b.Jitter)
    save_duration(2) = Factor * save_duration(1)

duration(3) = Jitter(save_duration(2), b.Jitter)
    save_duration = Factor * save_duration(2)
...
duration(n) = Jitter(save_duration(n-1), b.Jitter)

设置Backoff参数如下，按照上述公式得出的期望延时为[1,1.1)，[3,3.3), [9,9.9), [27,29.7)。实际运行如下，小数点一位后四舍五入得出实际延时为1.1, 3.3, 9.6, 28.2，与预期相符。 

 var DefaultRetry = wait.Backoff{
    Steps:    5,
    Duration: 1 * time.Second,
    Factor:   3,
    Jitter:   0.1,
}

结果：
2019-07-05 11:42:54.8779046 +0800 CST m=+0.135740401
2019-07-05 11:42:55.9399737 +0800 CST m=+1.197782901
2019-07-05 11:42:59.2240904 +0800 CST m=+4.481817401
2019-07-05 11:43:08.8232438 +0800 CST m=+14.080730501
2019-07-05 11:43:37.0058953 +0800 CST m=+42.262752301
timed out waiting for the condition

=> 最后看下Backoff.Cap的影响。设置Cap为10s，预期会比上面不带Cap的少执行2次(不带Cap限制的在Step为0时还会执行一次)。实际执行上也是如此 

 var DefaultRetry = wait.Backoff{
    Steps:    5,
    Duration: 1 * time.Second,
    Factor:   3,
    Jitter:   0.1,
    Cap:      time.Second*10,
}

结果：
2019-07-05 12:02:43.8678742 +0800 CST m=+0.120673901
2019-07-05 12:02:44.9294079 +0800 CST m=+1.182202101
2019-07-05 12:02:48.2125558 +0800 CST m=+4.465333301

ExponentialBackoff借鉴了TCP协议的指数退避算法，适用于可能会产生资源竞争的场景。指数退避可以有效地在没有缓存处理或缓存不足的场景下减小服务端的压力。

  • wait库的第二组 
 func  Poll (interval, timeout time.Duration, condition ConditionFunc) error 
func PollImmediate(interval, timeout time.Duration, condition ConditionFunc) error 
func PollInfinite(interval time.Duration, condition ConditionFunc) error 
func PollImmediateInfinite(interval time.Duration, condition ConditionFunc) error 
func PollUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error 
func PollImmediateUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error

Poll表示以interval的周期执行condition函数，直到timeout超时或condition返回true/err非空。

wait.Poll和wait.Until使用上还是有些类似的，区别在于一个使用timeout限制超时时间，一个使用chan提供主动停止调度。 

 import (
    "fmt"
    "k8s.io/apimachinery/pkg/util/wait"
    "time"
)


func main(){

    wait.Poll(time.Second, time.Second*5, func() (done bool, err error) {
        fmt.Println(time.Now())
        return false,nil
    })

结果：
2019-07-05 13:43:31.2622405 +0800 CST m=+1.069324901
2019-07-05 13:43:32.2619663 +0800 CST m=+2.069050701
2019-07-05 13:43:33.2626114 +0800 CST m=+3.069695801
2019-07-05 13:43:34.2626876 +0800 CST m=+4.069772001
2019-07-05 13:43:35.2624168 +0800 CST m=+5.069501201
2019-07-05 13:43:35.2624168 +0800 CST m=+5.069501201

PollInfinite相比Poll取消了timeout的限制。

PollUntil相比Until来说，PollUntil在condition函数返回true或error的时候会退出调度。

Poll和PollImmediate为一组，PollInfinite和PollImmediateInfinite为一组，PollUntil和PollImmediateUntil为一组，它们的细微差别在于前者在执行condition函数前会等待interval时间，后者则会首先运行condition函数，然后再检查是否需要等待(condition返回true或err非空时不会再等待)。如果不关注这点差异，用哪个都可以。

  • heap 堆(k8s.io/client-go/tools/cache)

实现heap需要实现下面Interface接口，heap使用队列实现了一个完全二叉树 

 // heap.Interface
type Interface interface {
    sort.Interface
    Push(x interface{}) // add x as element Len()
    Pop() interface{}   // remove and return element Len() - 1.
}

// sort.Interface
type Interface interface {
    // Len is the number of elements in the collection.
    Len() int
    // Less reports whether the element with
    // index i should sort before the element with index j.
    Less(i, j int) bool
    // Swap swaps the elements with indexes i and j.
    Swap(i, j int)
}

heap对外提供的方法为如下： 

 func Init(h Interface)
func Push(h Interface, x interface{})
func Pop(h Interface) interface{}
func Remove(h Interface, i int) interface{}
func Fix(h Interface, i int) // 当修改完队列中的index=i的元素后，重新排序

例子如下： 

 import (
    "container/heap"
    "fmt"
)
    
func GetAllHeapItems(t Heap_t,name string){
    items := []interface{}{}
    for t.Len() != 0{
        items = append(items, heap.Pop(&t))
    }
    fmt.Println(name,":",items)
}

type Heap_t []int
func (h Heap_t)Len() int{return len(h)}
func (h Heap_t)Less(i,j int)bool {return h[i]<h[j]}
func (h Heap_t)Swap(i,j int){h[i], h[j] = h[j], h[i]}
func (h *Heap_t)Push(x interface{}){*h = append(*h,x.(int))}
func (h *Heap_t)Pop() interface{}{
    if h.Len() == 0{
        return nil
    }
    x := (*h)[len(*h)-1]
    *h = (*h)[0:(len(*h) - 1)]
    return x
}

func main(){
    h := &Heap_t{4,2,6,80,100,45} //[1 2 4 8 80 45 6 23 56 100]
    heap.Init(h)
    GetAllHeapItems(*h,"h")

    h1 := &Heap_t{4,2,6,80,100,45}
    heap.Init(h1)
    h1.Push(3)
    GetAllHeapItems(*h1,"h1")

    h2 := &Heap_t{4,2,6,80,100,45}
    heap.Init(h2)
    GetAllHeapItems(*h2,"h2")

    h3 := &Heap_t{4,2,6,80,100,45}
    heap.Init(h3)
    (*h3)[2] = 200
    fmt.Println(1111,h3)
    heap.Fix(h3,2)
    fmt.Println(2222,h3)
    GetAllHeapItems(*h3,"h3")
}

结果：
h : [2 4 6 45 80 100]
h1 : [2 3 4 6 45 80 100]
h2 : [2 4 6 45 80 100]
1111 &[2 4 200 80 100 45]
2222 &[2 4 45 80 100 200]
h3 : [2 4 45 80 100 200]

heap的实现比较巧妙，使用队列实现了完全二叉树，比较适用于查询频繁的场景，原理解析可以参见这里

更多使用和例子参见官方文档

  • klog(k8s.io/klog) 实现执行日志打印
  • 使用select{}实现主协程不退出 
 func main(){
    ...
    select{}
}
  • 可以使用switch对地址进行判断 
 package main

import (
    "fmt"
)

func main(){
    type emptyCtx int
    background := new(emptyCtx)
    todo       := new(emptyCtx)
    typeSwitch := func (i interface{}) {
        switch i {
        case background:
            fmt.Println("background")
        case todo:
            fmt.Println("todo")
        default:
            fmt.Println("default")
        }
    }

    typeSwitch(background)
}

结果：
true
  • 限流 (“golang.org/x/time/rate”)

rate.Limiter使用令牌桶实现限流，它共有3组对外方法，多数场景下使用Wait，用于等待令牌。更多解析可以参见这里 

 func (lim *Limiter) Wait(ctx context.Context) (err error)
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)  
 func (lim *Limiter) Allow() bool
func (lim *Limiter) AllowN(now time.Time, n int) bool  
 func (lim *Limiter) Reserve() *Reservation
func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation

Limiter使用如下函数初始化，b为令牌桶大小，初始是满的；r为每秒往桶里面放token的速率 

 func NewLimiter(r Limit, b int) *Limiter

下面是Limiter的简单用法，其最终耗费时间为5s。计算方式为：

  • 需要处理20个事件由于桶一开始是满的，所以立即可以处理已有的10个token还剩下10个事件，此时桶已经空了，每秒往桶里面放token的速率为每秒2个，因此每秒可以处理2个事件，处理10个事件需要5秒，这就是5s的由来，即（20-b）/r

ps:NewLimiter的入参r可以大于b，但其实此时大于的部分并没有意义，受限于桶的大小，多余的token会被丢弃 

 import (
    "context"
    "fmt"
    "golang.org/x/time/rate"
    "time"
)

func main(){
    l := rate.NewLimiter(2, 10)
    ctx,cancel := context.WithCancel(context.Background())
    defer cancel()
    
    f:= func(ctx context.Context) error{
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:            
        }
        //do something
        return nil
    }

    start := time.Now()
    for i := 0; i < 20; i++ {
        err := l.Wait(ctx)
        if nil != err{
            fmt.Println(err)
            fmt.Println(time.Since(start))
            return
        }
        go f(ctx)
    }
    fmt.Println(time.Since(start))
}

结果：
5.0000404s

下例中，如果每秒处理的令牌小于2，调度频率为实际执行频率(每秒一次) 

 func main(){
    l := rate.NewLimiter(2, 10)
    ctx,cancel := context.WithCancel(context.Background())
    defer cancel()

    f:= func(ctx context.Context) error{
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        time.Sleep(1*time.Second)
        return nil
    }

    start := time.Now()
    for i := 0; i < 20; i++ {
        err := l.Wait(ctx)
        if nil != err{
            fmt.Println(err)
            fmt.Println(time.Since(start))
            return
        }
        f(ctx)
    }
    fmt.Println(time.Since(start))
}

结果：
20.0107691s

WaitN用于判断是否可以同时执行n个事件，每次消耗n个令牌。如下例子的总时间算法为：(5*6-10)/2=10 

 import (
    "context"
    "fmt"
    "golang.org/x/time/rate"
    "time"

)

func main(){
    l := rate.NewLimiter(2, 10)
    ctx,cancel := context.WithCancel(context.Background())
    defer cancel()

    f:= func(ctx context.Context) error{
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        return nil
    }

    start := time.Now()
    for i := 0; i < 6; i++ {
        err := l.WaitN(ctx,5)
        if nil != err{
            fmt.Println(err)
            fmt.Println(time.Since(start))
            return
        }
        f(ctx)
    }
    fmt.Println(time.Since(start))
}

结果：
10.0011304s

读取 yaml 文件

假设yaml文件内容如下，yaml文件中的每个首字母必须小写。PS:yaml文件中的所有字符建议都小写 

 scrapeInterval: 60
endpoints:
  - endpoint: cn-shanghai.log.aliyuncs.com
    accessKeyID: LTAI2KSu0MDauu2r
    accessKeySecret: D3m0j7vDmrAWf33SFUh3LJRF1QGgTu
    project: avacar-slb-sls
    logstore: avacar-sls
    consumerGroupName: endCursor
    consumerName: endConsumer1
    cursorPosition: END_CURSOR

在代码中定义对应的解析结构体，结构体成员首字母大写，注意每个元素后面的yaml对应的字符串需要与yaml文件中的元素对应，大小写一致。 

 type Config struct {
    ScrapeInterval int32
    Endpoints []Endpoints`yaml: "endpoints"`
}

type Endpoints struct {
    Endpoint string `yaml:"endpoint"`
    AccessKeyID string `yaml:"accessKeyID"`
    AccessKeySecret string `yaml:"accessKeySecret"`
    Project string `yaml:"project"`
    Logstore string `yaml:"logstore"`
    ConsumerGroupName string `yaml:"consumerGroupName"`
    ConsumerName string `yaml:"consumerName"`
    CursorPosition string `yaml:"cursorPosition"`
}

使用如下方式即可将yaml文件的内容提取出来，即config 

 var config Config
configContent, err := ioutil.ReadFile("D:\\test.yaml")
if err != nil {
    log.Panic("open file failed")
    return
}

yaml.Unmarshal(configContent,&config)
  • sync.pool

sync.Pool设计的目的是用来保存和复用临时对象，以减少内存分配，降低CG压力，在大量复用变量的场景下能显著提高运行效率 

 type S struct {
    num int
}

func BenchmarkWithPool(b *testing.B) {
    var s *S
    var pool = sync.Pool{
        New: func() interface{} { return new(S) },
    }
    for i := 0; i < b.N; i++ {
        for j := 0; j < 10000; j++ {
            s = pool.Get().(*S)
            s.num = 1
            s.num++
            pool.Put(s)
        }
    }
}

func BenchmarkWithNoPool(b *testing.B) {
    var s *S
    for i := 0; i < b.N; i++ {
        for j := 0; j < 10000; j++ {
            s = &S{num: 1}
            s.num++
        }
    }
}

func main(){
    t1 := time.Now().Nanosecond()
    BenchmarkWithPool(&testing.B{N:10})
    t2 := time.Now().Nanosecond() - t1
    fmt.Println("t2 =",t2)

    t3 := time.Now().Nanosecond()
    BenchmarkWithNoPool(&testing.B{N:10})
    t4 := time.Now().Nanosecond() - t3
    fmt.Println("t4 =",t4)}
}

结果：

t2 = 1992800t4 = 999000

从下面可以看出，put和get是按照顺序一对一的，如果get完，则调用New函数创建一个新的元素返回 

 // 建立对象
var pipe = &sync.Pool{New:func()interface{}{return "Hello,BeiJing"}}

// 放入
pipe.Put("Hello,World1")
pipe.Put("Hello,World2")
pipe.Put("Hello,World3")
// 取出
log.Println(pipe.Get())
log.Println(pipe.Get())
log.Println(pipe.Get())
// 再取就没有了,会自动调用NEW
log.Println(pipe.Get())

结果：

2019/12/02 15:24:47 Hello,World12019/12/02 15:24:47 Hello,World22019/12/02 15:24:47 Hello,World32019/12/02 15:24:47 Hello,BeiJing

参考：

作者： charlieroro

出处：

标签:
打赏 点赞(1)

文章来源：智云一二三科技

文章标题：client-go和golang源码中的技巧

文章地址：https://www.zhihuclub.com/86294.shtml

关于作者: 智云科技

相关文章

热门文章

1分享新浪图床上传接口源码

点赞(878)阅读(58,761)

2PHP简单实现路由Route功能

点赞(1.09K)阅读(35,925)

3Tideways、xhprof 和 xhgui 打造 PHP 非侵入式监控平台

点赞(305)阅读(4,263)

4centos系统如何查看是否安装了mysql

点赞(482)阅读(3,741)

5curl 工具简述

点赞(453)阅读(2,950)
网站地图