为什么顺序循环比Go中的并发方法运行得更快？

Question

场景: 我想快速读取一个大文本文件(lorem.txt例如下面的例子中有4.5毫米的行).我在下面的代码中尝试了三种不同的方法.

1. 非并发顺序处理
2. 频道和goroutines
3. Waitgroup和goroutines

伪基准: 这是我在下面运行时获得的典型输出; 下面的输出是快速和肮脏的时间增量,而不是完整的剖析/台架标记/测试.

sequential:     43.541828091 secs 4714074 lines
queued channel: 80.986544385 secs 4714074 lines
wait group:     260.200473751 secs 4712266 lines

问题: 为什么顺序循环比下面的其他两种方法更快？我错过了什么吗？

更新 我在更大的文本文件上运行示例代码(请参阅方案).我还为每个示例"重置"记录器,以防万一不这样做会在示例函数之间产生一些复合内存问题.还有其他人注意到,我的电脑是双核的,这可能是我的代码的许多问题之一.感谢所有反馈/答案.

package main

import (
    "bufio"
    "bytes"
    "fmt"
    "log"
    "os"
    "sync"
    "time"
)

var (
    textFile = "lorem.txt"
    buf      bytes.Buffer
    l        = log.New(&buf, "logger: ", log.Lshortfile)
    wg       sync.WaitGroup
    delta1   float64
    delta2   float64
    delta3   float64
    cnt1     = 0
    cnt2     = 0
    cnt3     = 0
)

func main() {

    // Wait Group Example
    exampleWaitGroup()

    // Queued Channel Example
    exampleQueuedChannel()

    // Sequential Loop Example
    exampleSequentialLoop()

    benchmarks := fmt.Sprintf("sequential:\t%v secs %v lines\nqueued channel:\t%v secs %v lines\nwait group:\t%v secs %v lines\n",
        delta1, cnt1,
        delta2, cnt2,
        delta3, cnt3,
    )

    fmt.Println(benchmarks)

}

func exampleSequentialLoop() {

    buf.Reset()
    l = log.New(&buf, "logger: ", log.Lshortfile)

    start := time.Now()

    file1, err := os.Open(textFile)
    if err != nil {
        log.Fatal(err)
    }

    defer file1.Close()

    scanner := bufio.NewScanner(file1)

    for scanner.Scan() {
        cnt1++
        l.Println(scanner.Text())
    }

    end := time.Now()
    delta1 = end.Sub(start).Seconds()

}

func exampleQueuedChannel() {

    buf.Reset()
    l = log.New(&buf, "logger: ", log.Lshortfile)

    start := time.Now()
    queue := make(chan string)
    done := make(chan bool)

    go processQueue(queue, done)

    file2, err := os.Open(textFile)
    if err != nil {
        log.Fatal(err)
    }

    defer file2.Close()

    scanner := bufio.NewScanner(file2)

    for scanner.Scan() {
        queue <- scanner.Text()
    }

    end := time.Now()
    delta2 = end.Sub(start).Seconds()
}

func exampleWaitGroup() {

    buf.Reset()
    l = log.New(&buf, "logger: ", log.Lshortfile)

    start := time.Now()

    file3, err := os.Open(textFile)
    if err != nil {
        log.Fatal(err)
    }

    defer file3.Close()

    scanner := bufio.NewScanner(file3)

    for scanner.Scan() {
        wg.Add(1)
        go func(line string) {
            defer wg.Done()
            l.Println(line)
            cnt3++
        }(scanner.Text())
    }

    wg.Wait()

    end := time.Now()
    delta3 = end.Sub(start).Seconds()

}

func processQueue(queue chan string, done chan bool) {
    for line := range queue {
        l.Println(line)
        cnt2++
    }
    done <- true
}

Answer 1

在exampleQueuedChannel你没有做任何事情并行.是的,你已经推出了另一个goroutine,但没有并行处理.原因是这queue是一个无缓冲的陈.当您写入无缓冲的chan时,编写器会阻塞,直到有人读取它为止.所以基本上你在写作时会阻塞,然后调度程序必须让goroutine进入睡眠状态,然后唤醒读取goroutine.然后那个人去睡觉了,作家醒来了.所以你在两个goroutines之间挣扎,调度程序正在进行大量训练.
如果你想在这里获得更好的性能,请使用缓冲的chan.如果你想要更高的性能,每个chan消息增加多个项目^{(有关频道性能影响的长期技术解释,请阅读此内容)}.

在exampleWaitGroup,你将为每一行推出一个新的goroutine.虽然推出新的goroutine并不昂贵,但它也不是免费的,而且对于调度程序来说也是更多的工作.defer也不是免费的.此外,您的记录器使用互斥锁,因此如果您的两个goroutine同时尝试登录,则一个将进入休眠状态,并且再次使用更多的调度程序.

您可以通过在分析器下启动代码并调查瓶颈所在的位置来自行调查这些问题.