在基于通道的管道中使用上下文超时

Question

我的设计利用了工作人员之间的通道管道。基本设计是将任务放置在扇出通道上，由长期以来喜欢的 gorouteans 充当工人，消耗来自通道的工作并处理它们。

+-------------+                        +--------+
| task source |  == fan out channels =>| worker |
+-------------+                        +--------+

每个工作人员 goroutean 都在进程的生命周期中存在，并将处理多个作业。

但是，由于新的要求，我现在必须实现超时以完成通过系统处理任务。我的理解是，跨函数的事情的超时通常是通过上下文来实现的。

但是我不确定如何在通过通道连接的工作人员之间传播上下文。

或者我问错了问题，他们是另一种更适合我要求的解决方案吗？

Answer 1

传入goroutine通道和上下文就可以了，像这样：

// ctx created earlier for all goroutines

for i := 0; i < workersCount; i++ {
    go worker(ctx, jobsChan)
}

所以现在您可以让所有工作人员超时。

但是可能您的超时与 1 个工作人员中的 1 个特定工作有关，与其他工作人员和工作无关，因此您需要为每个工作处理创建上下文，如下所示：

for i := 0; i < workersCount; i++ {
    go worker(jobsChan)
}

func worker(jobs <-chan Job) {
   // read from chan
   // and create ctx for job
   ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
   // ...
}

或者（如果我得到@Adrian 的观点正确）你可以这样做：

ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
job.ctx = ctx
job.ctxCancel = cancel
// put job int chan

for i := 0; i < workersCount; i++ {
    go worker(jobsChan)
}

func worker(jobs <-chan Job) {
   // read from chan
   // deal with job.ctx
   // ...
}