我使用調度程序，運行匿名函數：gocronj := Job{}j.ClientID = clientidj.Topic = topicj.Range = rangej.Interval = intervalj.Task = func(s string, t string){ fmt.Println(s) fmt.Println(t) Query_by_limit(s, t)// <-- data wanted}gocron.Every(j.Interval).Seconds().Do(j.Task, j.Topic, j.Range)gocron.Start()如何從每次到達時捕獲數據并將其發送以進行處理？Query_by_limit()
1 回答
小唯快跑啊
TA貢獻1863條經驗 獲得超2個贊
影響您要使用的解決方案的因素有很多。以下是在考慮如何繼續之前需要回答的問題。
您的圖書館如何運作？它是否同時執行作業，同時保持需要執行的作業的計劃順序？還是所有計劃作業都在同一線程/goroutine 中執行？
gocron您是否關心檢索到的數據以何種順序進行處理？是否需要按檢索順序處理它？
是要立即處理數據，還是也在單獨的計劃作業中處理數據？
數據處理所花費的時間是比兩個計劃作業之間的時間間隔的持續時間更多還是更少？
下面是一個解決方案，其假設是計劃程序不同時執行任務（某些庫以這種方式工作），數據處理永遠不會超過作業等待間隔的持續時間，并且需要按檢索順序處理數據。gocron
queryJob := Job{}
dataCh := make(chan interface{})
done := make(chan bool)
queryJob.ClientID = clientid
queryJob.Topic = topic
queryJob.Range = range
queryJob.Interval = interval
queryJob.Task = func(s string, t string, dataCh chan<- interface{}){
fmt.Println(s)
fmt.Println(t)
data, err := Query_by_limit(s, t)
if err != nil {
//handle error
}
dataCh <- data //if dataCh is an unbuffered channel, the job blocks here until the data is read in the data processing goroutine
}
//data processing goroutine
go func() {
for data := range dataCh {
//process data
}
done <-true
}()
gocron.Every(j.Interval).Seconds().Do(j.Task, j.Topic, j.Range, dataCh)
gocron.Start()
//when scheduler exits, you can close dataCh and give it a chance to be emptied
close(dataCh)
<-done
如果不想在數據處理 goroutine 中讀取數據之前阻止作業，則可以將通道設置為緩沖通道。dataCh
如果數據處理花費的時間比計劃作業的等待間隔長，并且您仍然希望保持需要處理的數據的順序，則解決方案會變得更加復雜。如果要在不同的計劃作業中處理數據，情況也是如此。在這些情況下，我建議嘗試使用隊列來存儲和維護其中數據順序的解決方案，以及在程序退出之前實現排出隊列并處理所有數據。
- 1 回答
- 0 關注
- 93 瀏覽
添加回答
舉報
0/150
提交
取消