go - 如何从 worker 那里订购结果，就好像没有使用 worker 一样？

Question

假设我有以下代码来读取行并将每行乘以 2，然后逐行打印出每一行。

我想使用 N 个 worker 。每个 worker 每次取 M 行并处理它们。更重要的是，我希望输出的打印顺序与输入的顺序相同。但是这里的示例并不能保证输出的打印顺序与输入的顺序相同。

https://gobyexample.com/worker-pools

以下 URL 还显示了一些示例。但我认为它们不符合我的要求。问题是输入可以任意长。在打印之前无法将所有内容保存在内存中。必须有一种方法可以从 worker 那里获得一些输出，可以确定 worker 的输出是否准备好打印然后打印。听起来应该有一个 master goroutine 来做这件事。但我不确定如何最有效地实现它，因为当 N 很大时，这个 master gorountine 很容易成为瓶颈。

How to collect values from N goroutines executed in a specific order?

谁能展示一个示例程序，该程序由 worker 按顺序生成并在可以打印时尽早打印结果？

$ cat main.go
#!/usr/bin/env gorun
// vim: set noexpandtab tabstop=2:

package main

import (
    "bufio"
    "fmt"
    "strconv"
    "io"
    "os"
    "log"
)

func main() {
    stdin := bufio.NewReader(os.Stdin)

    for {
        line, err := stdin.ReadString('\n')

        if err == io.EOF {
            if len(line) != 0 {
                i, _ := strconv.Atoi(line)
                fmt.Println(i*2)
            }
            break
        } else if err != nil {
            log.Fatal(err)
        }

        i, _ := strconv.Atoi(line[:(len(line)-1)])
        fmt.Println(i*2)
    }
}

Answer 1

如果 worker 知道他们的初始订单，例如例如，了解行号，然后让工作人员保留该信息(仅行号)。然后，您的员工将该信息反馈给您的结果 channel 。您的结果聚合代码从结果 channel 接收，然后在进一步处理(例如打印)之前根据初始订购信息对结果进行排序。

下面是您展示的其中一个示例的快速修改。

包主

import "fmt"
import "time"

type Result struct {
    Data, Seq int
}

type Job struct {
    Data string
    Seq  int
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for j := range jobs {
        fmt.Println("worker", id, "started  job", j)
        time.Sleep(time.Second)
        fmt.Println("worker", id, "finished job", j)
        results <- Result{len(j.Data), j.Seq}
    }
}

func main() {
    workload := 5

    jobs := make(chan Job, 100)
    results := make(chan Result, 100)

    output := make([]Result, workload)

    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    for j := 0; j < workload; j++ {
        jobs <- Job{ // explicit to make it clear
            Data: fmt.Sprintf("blah blah blah %d", j),
            Seq:  j,
        }
    }
    close(jobs)

    // receive results
    for a := 1; a <= workload; a++ {
        res := <-results
        output[res.Seq] = res

        // uncomment to see unordered
        // fmt.Printf("received: %#v", res)
    }

    for _, out := range output {
        fmt.Printf("output %#v\n", out)
    }
}

顺便说一句:如果您事先不知道您的工作量，这将无法正常工作...在这种情况下，您接收结果的代码在处理已经收到和订购的部分时需要更智能一些(作业):) .基本上等待第 0 行，然后等待下一个或打印已按顺序接收到的内容。

玩得开心!