k8s 中定时任务的实现

k8s 中有许多优秀的包都可以在平时的开发中借鉴与使用,比如,任务的定时轮询、高可用的实现、日志处理、缓存使用等都是独立的包,可以直接引用。本篇文章会介绍 k8s 中定时任务的实现,k8s 中定时任务都是通过 wait 包实现的,wait 包在 k8s 的多个组件中都有用到,以下是 wait 包在 kubelet 中的几处使用:

funcrun(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chanstruct{})(err error){ ...// kubelet 每5分钟一次从 apiserver 获取证书closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, clientConfig, clientCertificateManager,5*time.Minute)iferr !=nil{returnerr } closeAllConns, err := kubeletcertificate.UpdateTransport(wait.NeverStop, clientConfig, clientCertificateManager,5*time.Minute)iferr !=nil{returnerr } ...}...funcstartKubelet(k kubelet.Bootstrap, podCfg *config.PodConfig, kubeCfg *kubeletconfiginternal.KubeletConfiguration, kubeDeps *kubelet.Dependencies, enableServerbool){// 持续监听 pod 的变化gowait.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

一个示例:

packagemainimport("fmt""sync""time")funcmain(){varwg sync.WaitGroup timer1 := time.NewTimer(2* time.Second) ticker1 := time.NewTicker(2* time.Second) wg.Add(1)gofunc(t *time.Ticker){deferwg.Done()for{ <-t.C fmt.Println("exec ticker", time.Now().Format("2006年01月02日 15:04:05")) } }(ticker1) wg.Add(1)gofunc(t *time.Timer){deferwg.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):

funcJitterUntil(ffunc(),periodtime.Duration,jitterFactorfloat64,slidingbool,stopCh<-chanstruct{}) {vart *time.TimervarsawTimeoutboolfor{select{case<-stopCh:returndefault: } jitteredPeriod := periodifjitterFactor >0.0{ jitteredPeriod = Jitter(period, jitterFactor) }if!sliding { t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout) }func(){deferruntime.HandleCrash() f() }()ifsliding { t = resetOrReuseTimer(t, jitteredPeriod, sawTimeout) }select{case<-stopCh:returncase<-t.C: sawTimeout =true} }}...funcresetOrReuseTimer(t *time.Timer, d time.Duration, sawTimeoutbool)*time.Timer{ift ==nil{returntime.NewTimer(d) }if!t.Stop() && !sawTimeout { <-t.C } t.Reset(d)returnt}

几个关键点的说明:

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 中许多功能的实现也都是我们需要在平时工作中用的,其大部分包的性能都是经过大规模考验的,通过使用其相关的工具包不仅能学到大量的编程技巧也能避免自己造轮子。

欢迎学Java和大数据的朋友们加入java架构交流: 855835163

加群链接:https://jq.qq.com/?_wv=1027&k=5dPqXGI

群内提供免费的架构资料还有:Java工程化、高性能及分布式、高性能、深入浅出。高架构。性能调优、Spring,MyBatis,Netty源码分析和大数据等多个知识点高级进阶干货的免费直播讲解 可以进来一起学习交流哦

直播课堂地址:https://ke.qq.com/course/260263?flowToken=1007014

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传