Golang 非阻塞缓冲区

Question

同步示例：

type job struct {

    Id int

    Message string

}

for {

    // getJob() blocks until job is received

    job := getJob()

    doSomethingWithJob(job)

}

getJob我希望处理来自doSomethingWithJob. 例如，getJob 可以是从 MessagingQueue（如 RabbitMQ/Beanstalkd）或处理 HTTP 请求接收到的有效负载。

我不想阻止getJob，doSomethingWithJob反之亦然。但是，我确实想控制/缓冲作业的数量，以免系统过载。例如最大并发数为 5。

目前，goroutines 的概念让我感到困惑，所以任何正确方向的指针都将不胜感激，以帮助我学习。

更新：感谢@JimB 的帮助。为什么工人 5 总是接班？

jobCh := make(chan *job)

// Max 5 Workers

for i := 0; i < 5; i++ {

    go func() {

        for job := range jobCh {

            time.Sleep(time.Second * time.Duration(rand.Intn(3)))

            log.Println(i, string(job.Message))

        }

    }()

}

for {

    job, err := getJob()

    if err != nil {

        log.Println("Closing Channel")

        close(jobCh)

        break

    }

    jobCh <- job

}

log.Println("Complete")

示例输出

2016/06/09 22:19:57 5 {"id":10692,"name":"Test Message"}

2016/06/09 22:19:57 5 {"id":10687,"name":"Test Message"}

2016/06/09 22:19:57 5 {"id":10699,"name":"Test Message"}

2016/06/09 22:19:57 5 {"id":10701,"name":"Test Message"}

2016/06/09 22:19:57 5 {"id":10703,"name":"Test Message"}

2016/06/09 22:19:57 5 {"id":10704,"name":"Test Message"}

Answer 1

您可以启动 5 个从通道读取的 goroutine 来调用doSomethingWithJob. 这样，同时处理的作业永远不会超过 5 个。jobCh := make(chan *job)// start 5 workers to process jobsfor i := 0; i < 5; i++ {&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; for job := range jobCh {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; doSomethingWithJob(job)&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }()}// send jobs to workers as fast as we canfor {&nbsp; &nbsp; jobCh <- getJob()}