f
我已经编写了这个调度程序,但是当f
花费的时间超过输入recurring
时间间隔时,我无法让它“杀死”该输入函数。
如果那f
是一个进程而不是线程,那么我正在寻找的这个东西可能是某种定义的硬抢占。
这个f
定义是我无法控制的。它表示一个 ETL 作业,涉及在批处理执行期间处理来自多个数据库的数据。它f
已被写入go
并且工作正常,但我需要以某种方式对其进行某种控制,执行时间过长。
我知道f
是原子的,所以它要么在执行结束时更改数据库,要么不更改。所以在时间过长的情况下,可以认为“杀死”它是安全的。
func schedule(f func(), recurring time.Duration) chan struct{} {
ticker := time.NewTicker(recurring)
quit := make(chan struct{})
go func(inFunc func()) {
for {
select {
case <-ticker.C:
fmt.Println("Ticked")
// when "go" is removed, then if "f()" takes
// more than "recurring", then it postpones
// the following executions of "f()"
//
// instead somehow it should be "killed"
//
// check the timestamps in the execution of the test
go inFunc()
case <-quit:
fmt.Println("Stopping the scheduler")
ticker.Stop()
return
}
}
}(f)
return quit
}
为了看看发生了什么,我编写了这个测试:
func TestSlowExecutions(t *testing.T) {
// log some information using a human readable timestamp
dummyLog := func(format string, a ...interface{}) (n int, err error) {
prefix := fmt.Sprintf("[%v] ", time.Now())
message := fmt.Sprintf(format, a...)
return fmt.Printf("%s%s\n", prefix, message)
}
// UUID to be able to uniquely identify "fooFunc"
newUuid := func() string {
// sudo apt-get install uuid-runtime
uuid, _ := exec.Command("uuidgen").Output()
re := regexp.MustCompile(`\r?\n`)
uuidStr := re.ReplaceAllString(string(uuid), "")
return uuidStr
}
// simulate some sort of very slow execution
fooFunc := func() {
uuid := newUuid()
dummyLog("Ticked")
dummyLog("Starting task %s", uuid)
time.Sleep(2 * time.Second)
dummyLog("Finished task %s", uuid)
}
// test the very slow execution of "fooFunc"
quitChan := schedule(fooFunc, 1*time.Second)
time.Sleep(4 * time.Second)
close(quitChan)
// wait more to see the "closing" message
time.Sleep(4 * time.Second)
}