理解 Golang Worker Pattern

Source: Dev.to

启动工作者

intNumWorkers := 3

for w := 0; w < intNumWorkers; w++ {
    go InsertStudentData(ctx, tx, insertedStudentJobs, insertedStudentjobsErrCh)
}

func InsertStudentData(ctx context.Context, tx *sql.Tx, data <-chan models.StudentItem, errCh chan<- error) {
    // worker implementation
}

启动了三个 goroutine 工作者；每个工作者从 StudentItem 值的通道中读取数据，并通过 errCh 上报错误。

向工作者发布任务/作业

numJobs := len(StudentItems)
jobs := make(chan models.StudentItem, numJobs)

for i := 0; i < numJobs; i++ {
    jobs <- StudentItems[i]
}
close(jobs)

func InsertStudentWorker(ctx context.Context, tx *sql.Tx, data <-chan models.StudentItem, errCh chan<- error) {
    for d := range data {
        err := repository.InsertStudentItemFromWorker(ctx, tx, d)
        // handle err if needed
    }
}

调用方用所有 StudentItem 作业填充一个带缓冲的通道，然后关闭它。工作者从只读通道 (data <-chan models.StudentItem) 中获取作业并并发处理。

收集结果

numJobs := len(StudentItems)
jobs := make(chan models.StudentItem, numJobs)

for i := 0; i < numJobs; i++ {
    jobs <- StudentItems[i]
}
close(jobs)

// read from the error channel
for j := 0; j < numJobs; j++ {
    err := <-jobsErrCh
    if err != nil {
        _ = tx.Rollback()
        return err
    }
}

func InsertStudentWorker(ctx context.Context, tx *sql.Tx, data <-chan models.StudentItem, errCh chan<- error) {
    for d := range data {
        err := repository.InsertStudentItemFromWorker(ctx, tx, d)
        errCh <- err
    }
}

每个工作者将其结果（或错误）发送到只写通道 (errCh)。由于调用方知道已分发的作业数量，它会读取恰好相同数量的结果，处理任何错误，而无需了解是哪一个具体工作者产生的。