k8s 中定時任務的實現
k8s 中有許多優秀的包都可以在平時的開發中借鑑與使用，比如，任務的定時輪詢、高可用的實現、日誌處理、快取使用等都是獨立的包，可以直接引用。本篇文章會介紹 k8s 中定時任務的實現，k8s 中定時任務都是通過 wait 包實現的，wait 包在 k8s 的多個元件中都有用到，以下是 wait 包在 kubelet 中的幾處使用：
func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan struct{}) (err error) {
...
// kubelet 每5分鐘一次從 apiserver 獲取證書
closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, clientConfig, clientCertificateManager, 5*time.Minute)
if err != nil {
return err
}
closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, clientConfig, clientCertificateManager, 5*time.Minute)
if err != nil {
return err
}
...
}
...
func startKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration,kubeDeps *kubelet.Dependencies, enableServer bool) {
// 持續監聽 pod 的變化
go wait.Until(func() {
k.Run(podCfg.Updates())
}, 0, wait.NeverStop)
...
}
golang 中可以通過 time.Ticker 實現定時任務的執行，但在 k8s 中用了更原生的方式，使用 time.Timer 實現的。time.Ticker 和 time.Timer 的使用區別如下：
func (t *Timer) Reset(d Duration) bool
一個示例：
package main
import (
"fmt"
"sync"
"time"
)
func main() {
var wg sync.WaitGroup
timer1 := time.NewTimer(2 * time.Second)
ticker1 := time.NewTicker(2 * time.Second)
wg.Add(1)
go func(t *time.Ticker) {
defer wg.Done()
for {
<-t.C
fmt.Println("exec ticker", time.Now().Format("2006-01-02 15:04:05"))
}
}(ticker1)
wg.Add(1)
go func(t *time.Timer) {
defer wg.Done()
for {
<-t.C
fmt.Println("exec timer", time.Now().Format("2006-01-02 15:04:05"))
t.Reset(2 * time.Second)
}
}(timer1)
wg.Wait()
}
一、wait 包中的核心程式碼
核心程式碼（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)
}
select {
case <-stopCh:
return
case <-t.C:
sawTimeout = true
}
}
}
...
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
}
幾個關鍵點的說明：
- 1、如果 sliding 為 true，則在 f() 執行之後計算週期。如果為 false，那麼 period 包含 f() 的執行時間。
- 2、在 golang 中 select 沒有優先順序選擇，為了避免額外執行 f(),在每次迴圈開始後會先判斷 stopCh chan。
k8s 中 wait 包其實是對 time.Timer 做了一層封裝實現。
二、wait 包常用的方法
1、定期執行一個函式，永不停止，可以使用 Forever 方法：
func Forever(f func(), period time.Duration)
2、在需要的時候停止迴圈，那麼可以使用下面的方法，增加一個用於停止的 chan 即可，方法定義如下：
func Until(f func(), period time.Duration, stopCh <-chan struct{})
上面的第三個引數 stopCh 就是用於退出無限迴圈的標誌，停止的時候我們 close 掉這個 chan 就可以了。
3、有時候，我們還會需要在執行前去檢查先決條件，在條件滿足的時候才去執行某一任務，這時候可以使用 Poll 方法：
func Poll(interval, timeout time.Duration, condition ConditionFunc)
這個函式會以 interval 為間隔，不斷去檢查 condition 條件是否為真，如果為真則可以繼續後續處理；如果指定了 timeout 引數，則該函式也可以只常識指定的時間。
4、PollUntil 方法和上面的類似，但是沒有 timeout 引數，多了一個 stopCh 引數，如下所示：
PollUntil(interval time.Duration, condition ConditionFunc, stopCh <-chan struct{}) error
此外還有 PollImmediate 、 PollInfinite 和 PollImmediateInfinite 方法。
三、總結
本篇文章主要講了 k8s 中定時任務的實現與對應包（wait）中方法的使用。通過閱讀 k8s 的原始碼，可以發現 k8s 中許多功能的實現也都是我們需要在平時工作中用的，其大部分包的效能都是經過大規模考驗的，通過使用其相關的工具包不僅能學到大量的程式設計技巧也能避免自己造輪子。
參考：http://liubin.org/blog/2018/04/27/learn-from-k8s-running-func-periodically/