背景
- 在平台开发过程中，当有些执行完某些操作后，可能还需要等待直到某个条件满足的时候
- 比如在CMDB删除了一个项目，CMDB需要一直等待其他系统资源回收完，才能在自身彻底释放对应的资源， 那就需要一直在哪里等待 今天主要是一个关于定时轮训、等待的工具方法实现
完整代码
package waitimport ("errors""math/rand""time"
)// Jitter 允许时间在一定范围内的波动
func Jitter(duration time.Duration, maxFactor float64) time.Duration {if maxFactor <= 0.0 {maxFactor = 1.0}wait := duration + time.Duration(rand.Float64()*maxFactor*float64(duration))return wait
}// resetReuseTimer 重置timer
func resetReuseTimer(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
}// HandlerCrash 控制失败异常
func HandlerCrash() {println("crash")
}// JitterUtil 周期调度函数执行
func JitterUtil(f func(), period time.Duration, jitterFactor float64, sliding bool, stopCh <-chan struct{}) {var t *time.Timervar sawTimeout boolfor {select {case <-stopCh:returndefault:}jitteredPeriod := periodif jitterFactor > 0.0 {// 使用jitterFactor可以生成一个范围内随机的time.DurationjitteredPeriod = Jitter(jitteredPeriod, jitterFactor)}if !sliding {t = resetReuseTimer(t, jitteredPeriod, sawTimeout)}func() {defer HandlerCrash()f()}()if sliding {t = resetReuseTimer(t, jitteredPeriod, sawTimeout)}select {case <-stopCh:returncase <-t.C:sawTimeout = true}}
}// Util 是封装的JitterUtil并提供了jitterFactor和sliding的默认值
func Util(f func(), period time.Duration, stopCh <-chan struct{}) {JitterUtil(f, period, 0.0, true, stopCh)
}// WaiterFunc 等待某个函数执行
type WaitFunc func(done <-chan struct{}) <-chan struct{}// ConditionFunc 等待函数条件呗满足
type ConditionFunc func() (done bool, err error)var ErrWaitTimeout = errors.New("timed out waiting for the condition")// WaitFor 执行某个函数活着某个条件呗满足
func WaitFor(wait WaitFunc, fn ConditionFunc, done <-chan struct{}) error {c := wait(done)for {_, open := <-cok, err := fn()if err != nil {return err}if ok {return nil}if !open {break}}return ErrWaitTimeout
}func poller(interval, timeout time.Duration) WaitFunc {return WaitFunc(func(done <-chan struct{}) <-chan struct{} {ch := make(chan struct{})go func() {defer close(ch)tick := time.NewTicker(interval)defer tick.Stop()var after <-chan time.Timeif timeout != 0 {timer := time.NewTimer(timeout)after = timer.Cdefer timer.Stop()}for {select {case <-tick.C:select {case ch <- struct{}{}:default:}case <-after:returncase <-done:return}}}()return ch})
}// pollInternal
func pollInternal(wait WaitFunc, condition ConditionFunc) error {done := make(chan struct{})defer close(done)return WaitFor(wait, condition, done)
}// Poll 提供对外统一的接口
func Poll(interval, timeout time.Duration, condition ConditionFunc) error {return pollInternal(poller(interval, timeout), condition)
}
复制代码代码都比较简单， 但在K8s里面很多资源更新方法里面，都使用了这些函数, 比如JOB的更新里面
// UpdateJobFunc updates the job object. It retries if there is a conflict, throw out error if
// there is any other errors. name is the job name, updateFn is the function updating the
// job object.
func UpdateJobFunc(c clientset.Interface, ns, name string, updateFn func(job *batch.Job)) {ExpectNoError(wait.Poll(time.Millisecond*500, time.Second*30, func() (bool, error) {job, err := GetJob(c, ns, name)if err != nil {return false, fmt.Errorf("failed to get pod %q: %v", name, err)}updateFn(job)_, err = UpdateJob(c, ns, job)if err == nil {Logf("Successfully updated job %q", name)return true, nil}if errors.IsConflict(err) {Logf("Conflicting update to job %q, re-get and re-update: %v", name, err)return false, nil}return false, fmt.Errorf("failed to update job %q: %v", name, err)}))
}
复制代码今天写的比较简单，都是晚上回来看的，白天还药做平台开发，今天就这样吧
