golang的串行处理和并行处理区别

Mathilda ·

更新时间:2024-09-21

· 1662 次阅读

程序运算时往往需要数据，而数据的IO又往往需要时间传输，而常见的串行处理，是一个任务处理完成才接着处理新的任务，其效率低下可想而知。假如有3个任务要处理，而每个任务都有一定的阻塞，串行处理大概是这个样子:

main.go


package main
import (
    "fmt"
    "time"
)
type Task struct {
    Duration time.Duration
    Name string
}
func main() {
    // 声明要处理的任务
    taskList := []Task {
        {
            1,
            "处理1",
        },
        {
            2,
            "处理2",
        },
        {
            3,
            "处理3",
        },
    }
    starTime := time.Now().Unix()
    for _, item := range taskList {
        goProcess(item)
    }
    fmt.Printf("用时:%ds\n", time.Now().Unix() - starTime)
}
/**
 * 进行处理
 */
func goProcess(task Task) {
    time.Sleep(time.Second * task.Duration) // 假设这是io阻塞会在这里阻塞一些时间，而这些io可能是网络io也可能是磁盘io等，io完成后才会接着运行下去
    fmt.Printf("任务:%s 处理完毕\n", task.Name)
}

处理打印结果
任务:处理1 处理完毕
任务:处理2 处理完毕
任务:处理3 处理完毕
用时:6s

这种串行处理遇到IO阻塞时，弊端就是太费时间了

可以看到，程序在计算时或许用不到多少时间反而是IO阻塞占用了大量的时间。这种占着茅坑不拉屎，外面等着人直跳脚，确实不太好。而用异步处理则可以跳过阻塞，达到避免占坑的情况发生。

用协程的话，则可以在阻塞时先异步执行下去而不用等待，等所有协程都处理结束，再把处理的结果汇总起来就可以了，代码大概是这样子：

main.go


package main
import (
    "fmt"
    "sync"
    "time"
)
type Task struct {
    Duration time.Duration
    Name string
}
func main() {
    // 声明要处理的任务
    taskList := []Task {
        {
            1,
            "处理1",
        },
        {
            2,
            "处理2",
        },
        {
            3,
            "处理3",
        },
    }
    starTime := time.Now().Unix()
    var res []string //处理结果收集
    resChang := make(chan string, len(taskList))
    wg := &sync.WaitGroup{}
    // 这里收集异步处理的结果, 通过管道把数据传递过来，类似于单一订阅功能吧
    go func() {
        wg.Add(1)
        defer wg.Done() // 通道关闭后 处理结果也收集完毕，则触发 用于通知下方批处理，处理结果已经收集完毕
        for resItem := range resChang {
            res = append(res, resItem)
        }
    }()
    taskWG := &sync.WaitGroup{}
    for _, item := range taskList {
        taskWG.Add(1) // 批处理 信号量+1
        go goProcess(item, &resChang, taskWG)
    }
    taskWG.Wait()// 这里阻塞，等待所有处理执行完毕, 才接着运行下去
    close(resChang)// 已经处理完毕后就关闭处理传输通道
    wg.Wait() // 这是阻塞 等待处理收集完毕, 才接着运行去
    // 打印批处理收集的处理结果
    for _, i := range res {
        fmt.Printf("%s", i)
    }
    fmt.Printf("用时:%ds\n", time.Now().Unix() - starTime)
}
/**
 * 进行处理
 */
func goProcess(task Task, resChan *chan string, taskWG *sync.WaitGroup) {
    time.Sleep(time.Second * task.Duration) // 假设这是io阻塞会在这里阻塞一些时间，而这些io可能是网络io也可能是磁盘io等，才会接着运行下去
    res := fmt.Sprintf("任务:%s 处理完毕\n", task.Name)
    defer func() {
        *resChan <- res // 把处理结果传出去
        taskWG.Done() // 批处理信号量-1 来报告处理完毕
    }()
}