代码之家 › 专栏 › 技术社区 › kentor

从输入通道正确批处理项目

go mysql

kentor · 技术社区 · 7 年前

用例

我想在MySQL数据库中保存大量数据,通过通道接收这些数据。出于性能原因,我将它们分为10个项目进行批量处理。我仅每3小时收到一次输入项。

假设我得到了10004个项目,剩下4个项目,因为我的go例程等待10个项目,然后在批处理中“刷新它们”。我想确保它创建了一个少于10个项目的批次,以防该频道中没有更多的项目(该频道也被制作人关闭)。

代码:

// ProcessAudits sends the given audits in batches to SQL
func 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{}
        }
    }
}

2 回复 | 直到 7 年前

Maxim 7 年前

你也可以使用定时器。请在这里使用示例 https://play.golang.org/p/0atlGVCL-px

a-h 7 年前

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

推荐文章

ginad · 如何在go-app组件中执行javascript代码

1 年前

Jason Fingar · 方法在另一个方法的上下文中不可访问

1 年前

Flissi Hamed · 从抓取aliexpress到使用chromedp的无头浏览器

1 年前

Adam Presley · GORM错误:“运算符不存在:bigint=text”

1 年前

Jess The Witch · GCP云功能中处理延迟任务的模式

1 年前

Moonlit · 在Golang中将`func()Foo`转换为`func)any`?

1 年前

pmoubed · 这是使用计时器的goroutine泄漏吗?

1 年前

Kurt Peek · 在Go模板中,如何仅在字符串值非空的情况下添加一行?

1 年前

Harimbola Santatra · 如何获取JSON中包含unicode代码点的键的值?

1 年前

techStud · 为什么Go在变量超出作用域后立即取消引用悬挂指针时不会抛出错误?

1 年前