如何对workers的结果进行排序，就像没有使用workers一样？

Question

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

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

https://gobyexample.com/worker-pools

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

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

有没有人可以展示一个示例程序，按顺序显示工人的结果，并在可以打印的情况下尽早打印结果？

$ 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

如果工人知道初始订单，例如，他们被告知行号，那么让工人保留该信息(只保留行号)。然后，您的员工将该信息反馈给您的结果频道。您的结果聚合代码从results通道接收结果，然后在进一步处理(例如打印)之前根据初始排序信息对结果进行排序。

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

 package main

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行，然后等待下一行，或者按顺序打印已经接收到的内容。

玩得开心!