concurrency - 戈朗 : Producer/Consumer concurrency model but with serialized results

Question

func main() {
    jobs := []Job{job1, job2, job3}
    numOfJobs := len(jobs)
    resultsChan := make(chan *Result, numOfJobs)
    jobChan := make(chan *job, numOfJobs)
    go consume(numOfJobs, jobChan, resultsChan)
    for i := 0; i < numOfJobs; i++ {
        jobChan <- jobs[i]
    }
    close(jobChan)

    for i := 0; i < numOfJobs; i++ {
        <-resultsChan
    }
    close(resultsChan)
}

func (b *Blockchain) consume(num int, jobChan chan *Job, resultsChan chan *Result) {
    for i := 0; i < num; i++ {
        go func() {
            job := <-jobChan
            resultsChan <- doJob(job)
        }()
    }
}

在上面的示例中，作业被推送到 jobChan 中，goroutines 将其从 jobChan 中拉出并并发执行作业并将结果推送到 resultsChan 中。然后我们将从 resultsChan 中提取结果。

问题一:

在我的代码中，没有序列化/线性化的结果。虽然jobs的顺序是job1, job2, job3。结果可能会显示为 job3、job1、job2，具体取决于哪个花费的时间最长。

我仍然希望同时执行这些作业，但是，我需要确保结果从 resultsChan 中以与作为作业进入时相同的顺序出现。

问题2:

我有大约 30 万个作业，这意味着代码将生成多达 30 万个 goroutine。拥有如此多的 goroutine 是否有效，或者我最好将这些作业分成 100 个左右的片段，让每个 goroutine 处理 100 个而不是 1 个。

Answer 1

这是我处理序列化的一种方式(并且还设置了有限数量的工作人员)。我设置了一些带有输入和输出字段以及同步 channel 的工作对象，然后我循环遍历它们，挑选它们完成的所有工作并给它们一个新的工作。然后我最后一次通过它们以拾取所有遗留下来的已完成工作。请注意，您可能希望工作人员数量稍微超过您的核心数量，这样即使有一个异常长的工作，您也可以让所有资源保持忙碌一段时间。代码位于 http://play.golang.org/p/PM9y4ieMxw及以下。

这是毛茸茸的(比我记得在坐下来写一个例子之前毛茸茸的!)——很想看看其他人有什么，要么只是更好的实现，要么是一种完全不同的方式来实现你的目标。

package main

import (
    "fmt"
    "math/rand"
    "runtime"
    "time"
)

type Worker struct {
    in     int
    out    int
    inited bool

    jobReady chan bool
    done     chan bool
}

func (w *Worker) work() {
    time.Sleep(time.Duration(rand.Float32() * float32(time.Second)))
    w.out = w.in + 1000
}
func (w *Worker) listen() {
    for <-w.jobReady {
        w.work()
        w.done <- true
    }
}
func doSerialJobs(in chan int, out chan int) {
    concurrency := 23
    workers := make([]Worker, concurrency)
    i := 0
    // feed in and get out items
    for workItem := range in {
        w := &workers[i%
            concurrency]
        if w.inited {
            <-w.done
            out <- w.out
        } else {
            w.jobReady = make(chan bool)
            w.done = make(chan bool)
            w.inited = true
            go w.listen()
        }
        w.in = workItem
        w.jobReady <- true
        i++
    }
    // get out any job results left over after we ran out of input
    for n := 0; n < concurrency; n++ {
        w := &workers[i%concurrency]
        if w.inited {
            <-w.done
            out <- w.out
        }
        close(w.jobReady)
        i++
    }
    close(out)
}
func main() {
    runtime.GOMAXPROCS(10)
    in, out := make(chan int), make(chan int)
    allFinished := make(chan bool)
    go doSerialJobs(in, out)
    go func() {
        for result := range out {
            fmt.Println(result)
        }
        allFinished <- true
    }()
    for i := 0; i < 100; i++ {
        in <- i
    }
    close(in)
    <-allFinished
}

请注意，此示例中只有 in 和 out 携带实际数据——所有其他 channel 仅用于同步。