我使用調(diào)度程序，運行匿名函數(shù)：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()如何從每次到達時捕獲數(shù)據(jù)并將其發(fā)送以進行處理？Query_by_limit()
1 回答
小唯快跑啊
TA貢獻1863條經(jīng)驗 獲得超2個贊
影響您要使用的解決方案的因素有很多。以下是在考慮如何繼續(xù)之前需要回答的問題。
您的圖書館如何運作？它是否同時執(zhí)行作業(yè)，同時保持需要執(zhí)行的作業(yè)的計劃順序？還是所有計劃作業(yè)都在同一線程/goroutine 中執(zhí)行？
gocron您是否關(guān)心檢索到的數(shù)據(jù)以何種順序進行處理？是否需要按檢索順序處理它？
是要立即處理數(shù)據(jù)，還是也在單獨的計劃作業(yè)中處理數(shù)據(jù)？
數(shù)據(jù)處理所花費的時間是比兩個計劃作業(yè)之間的時間間隔的持續(xù)時間更多還是更少？
下面是一個解決方案，其假設(shè)是計劃程序不同時執(zhí)行任務(wù)（某些庫以這種方式工作），數(shù)據(jù)處理永遠不會超過作業(yè)等待間隔的持續(xù)時間，并且需要按檢索順序處理數(shù)據(jù)。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
如果不想在數(shù)據(jù)處理 goroutine 中讀取數(shù)據(jù)之前阻止作業(yè)，則可以將通道設(shè)置為緩沖通道。dataCh
如果數(shù)據(jù)處理花費的時間比計劃作業(yè)的等待間隔長，并且您仍然希望保持需要處理的數(shù)據(jù)的順序，則解決方案會變得更加復雜。如果要在不同的計劃作業(yè)中處理數(shù)據(jù)，情況也是如此。在這些情況下，我建議嘗試使用隊列來存儲和維護其中數(shù)據(jù)順序的解決方案，以及在程序退出之前實現(xiàn)排出隊列并處理所有數(shù)據(jù)。
- 1 回答
- 0 關(guān)注
- 105 瀏覽
添加回答
舉報
0/150
提交
取消