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

Question

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

id content last_id create_time

Answer 1

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

Main：

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))
}

go游乐场上的代码：https://play.golang.org/p/X9Q991h6ru_n