用例我想在通過通道接收的 MySQL 數(shù)據(jù)庫中保存大量數(shù)據(jù)。出于性能原因，我以 10 件為一組處理它們。我每 3 小時(shí)才收到一次輸入項(xiàng)目。問題假設(shè)我得到 10004 個(gè)項(xiàng)目，將剩下 4 個(gè)項(xiàng)目，因?yàn)槲业?go 例程在批量“沖走”之前等待 10 個(gè)項(xiàng)目。我想確保它創(chuàng)建一個(gè)少于 10 個(gè)項(xiàng)目的批次，以防該通道中沒有更多項(xiàng)目（然后生產(chǎn)者也關(guān)閉了通道）。代碼：// ProcessAudits sends the given audits in batches to SQLfunc ProcessAudits(done <-chan bq.Audit) { var audits []bq.Audit for auditRow := range done { user := auditRow.UserID.StringVal log.Infof("Received audit %s", user) audits = append(audits, auditRow) if len(audits) == 10 { upsertBigQueryAudits(audits) audits = []bigquery.Audit{} } }}我是 Go 的新手，我不確定如何正確實(shí)施它？
2 回答
紅顏莎娜
TA貢獻(xiàn)1842條經(jīng)驗(yàn) 獲得超13個(gè)贊
這是一個(gè)工作示例。當(dāng)通道關(guān)閉時(shí)，范圍退出，因此您可以在循環(huán)后處理任何剩余的項(xiàng)目。
package main
import (
"fmt"
"sync"
)
type Audit struct {
ID int
}
func upsertBigQueryAudits(audits []Audit) {
fmt.Printf("Processing batch of %d\n", len(audits))
for _, a := range audits {
fmt.Printf("%d ", a.ID)
}
fmt.Println()
}
func processAudits(audits <-chan Audit, batchSize int) {
var batch []Audit
for audit := range audits {
batch = append(batch, audit)
if len(batch) == batchSize {
upsertBigQueryAudits(batch)
batch = []Audit{}
}
}
if len(batch) > 0 {
upsertBigQueryAudits(batch)
}
}
func produceAudits(x int, to chan Audit) {
for i := 0; i < x; i++ {
to <- Audit{
ID: i,
}
}
}
const batchSize = 10
func main() {
var wg sync.WaitGroup
audits := make(chan Audit)
wg.Add(1)
go func() {
defer wg.Done()
processAudits(audits, batchSize)
}()
wg.Add(1)
go func() {
defer wg.Done()
produceAudits(25, audits)
close(audits)
}()
wg.Wait()
fmt.Println("Complete")
}
輸出：
Processing batch of 10
0 1 2 3 4 5 6 7 8 9
Processing batch of 10
10 11 12 13 14 15 16 17 18 19
Processing batch of 5
20 21 22 23 24
Complete
拉莫斯之舞
TA貢獻(xiàn)1820條經(jīng)驗(yàn) 獲得超10個(gè)贊
您也可以使用定時(shí)器。在這里玩例子https://play.golang.org/p/0atlGVCL-px
func printItems(items []int) {
fmt.Println(items)
return
}
func main() {
items := []int {1,2,3,4,5,6,7,8}
ch := make(chan int, 5)
go func(ch <-chan int) {
timer := time.NewTimer(1 * time.Second)
temp := make([]int, 0, 5)
for {
select {
case i := <- ch:
timer.Reset(1 * time.Second)
temp = append(temp, i)
if len(temp) == 5 {
printItems(temp)
temp = []int {}
}
case <- timer.C:
printItems(temp)
temp = []int {}
}
}
}(ch)
for k, i := range items {
fmt.Println("Send ", i)
ch <- i
if k == 7 {
time.Sleep(5 * time.Second)
}
}
}
- 2 回答
- 0 關(guān)注
- 162 瀏覽
添加回答
舉報(bào)
0/150
提交
取消