如何使用goroutine和channel批量读取mysql数据答案

【问题标题】：how to read mysql data using goroutine and channel in bulk如何使用goroutine和channel批量读取mysql数据
【发布时间】：2021-08-30 22:42:47
【问题描述】：

我是golang的新手，现在需要在mysql中读取大量数据，所以我想使用goroutine和channel来高性能获取数据，但不知道如何避免每个goroutine的数据重复和使整个过程稳定。例如，表模式如下，我想获取所有create_time小于1000000000000000000的记录，我想创建10个goroutine并同时读取数据，每个goroutine做一些业务逻辑，如何设计代码？谢谢你

id content last_id create_time

【问题讨论】：

创建 N 个 goroutines 都监听一个频道。在一个循环中，逐一读取 DB 行并将它们发送到通道。每个 goroutine 处理一个记录。您需要提供更多细节来决定如何处理计算结果。
我改变了我的问题并给出了我的要求的大致架构，可以帮助审查，谢谢
一次读取 1 条记录在计算上是不可行的。您如何指出一次读取一行的解决方案@BurakSerdar
@advayrajhansa，您一次从数据库中获取记录，对吧？
如果能读多个就更好，但只能读一个也可以

标签： go goroutine

【解决方案1】：

我建议您创建一个将数据发布到您的频道的 goroutine。然后你可以添加监听器去例程来处理发布的数据。这可以通过以下方式完成：

主要：

const GoroutineCount = 10

type SomeData []int

func main() {
    ch := make(chan SomeData, 1)

    go PublishData(ch)

    for i := 0; i < GoroutineCount; i++ {
        go ProcessData(ch)
    }
}

作为假设，我使用了一个简单的 int 切片作为数据。这可以是任何需要的结构的切片。

将数据发布到频道：

const ChunkSize = 1000

func PublishData(ch chan SomeData) {
    // Assume having 10000 records in result set
    // This has to come from db transaction
    res := make([]int, 10000)

    // split into chunks of 1000
    chunks := GetChunk(res)

    // write chunk data to channel
    for i := range chunks {
        ch <- chunks[i]
    }
}

func GetChunk(input SomeData) []SomeData {
    var result []SomeData

    boundary := len(input)
    index := 0
    for index = 0; boundary >= ChunkSize; index+=ChunkSize {
        boundary -= ChunkSize
        lastIndex := index+ChunkSize
        result = append(result, input[index:lastIndex])
    }
    boundary = len(input) % ChunkSize
    if boundary > 0 {
        lastIndex := index+ boundary
        result = append(result, input[index:lastIndex])
    }

    return result
}

将单个块处理为：

func ProcessData(ch chan SomeData) {
    // Read single chunk
    res := <-ch

    // Process chunk data
    fmt.Printf("len %d\n", len(res))
}

代码在操场上：https://play.golang.org/p/X9Q991h6ru_n

【讨论】：