上一章能查询一个商品价格。现在增加一项运费，book 的 2100 分加上运费 500 分，总额为 2600 分。接下来还要一次处理多个商品，并在失败时停止剩余工作。
每个阶段都有业务层 quote.go 与执行层 main.go。业务层保留 Price、PriceSource 和 Service，执行层用 Task 表达一项工作。这样可以分别研究报价规则和并发安排。
05：让两项独立查询同时开始
先看两项任务如何启动与交回结果：
learning/go-patterns/stages/05-parallel/main.go · L1–L51 · 打开完整源码
// Stage 05: start two independent tasks concurrently and collect both results.
package main
import (
"context"
"errors"
"fmt"
)
// Task is one unit of work. The context is available for a caller that needs
// to pass request-scoped state or observe cancellation in a later stage.
type Task func(context.Context) (string, error)
type indexedResult struct {
index int
value string
err error
}
// RunTwo starts both tasks before it waits for either result. Results are put
// back in input order, even when the second task finishes first. Both tasks
// are always collected; fail-fast cancellation is the subject of stage 07.
func RunTwo(ctx context.Context, first, second Task) ([2]string, error) {
var values [2]string
if ctx == nil {
return values, errors.New("context must not be nil")
}
if first == nil || second == nil {
return values, errors.New("both tasks are required")
}
tasks := [...]Task{first, second}
results := make(chan indexedResult, len(tasks))
for index, task := range tasks {
go func(index int, task Task) {
value, err := task(ctx)
results <- indexedResult{index: index, value: value, err: err}
}(index, task)
}
var firstErr error
for range tasks {
result := <-results
values[result.index] = result.value
if result.err != nil && firstErr == nil {
firstErr = result.err
}
}
return values, firstErr
}这段代码可以按四个动作读：
- 分配容量为二的结果 channel。
- 启动两个 goroutine，各执行一项任务。
- 每项任务发送带下标的结果。
- 调用方收齐两项，再按输入顺序返回。
任务可能先后完成，但返回值的位置保持稳定。缓冲容量足以容纳两次发送；接收方仍然会读完两项，失败时也遵守同一份收尾约定。
这里没有关闭结果 channel，因为接收次数已经由两项任务确定。channel 在整个流程结束、失去引用后由 Go 管理。需要用 range 持续接收直到结束的流程，才需要额外设计关闭通知。
接回查价业务
learning/go-patterns/stages/05-parallel/quote.go · L1–L51 · 打开完整源码
package main
import (
"context"
"errors"
"fmt"
)
var ErrUnknownItem = errors.New("unknown item")
type Price struct {
Item string
Cents int
}
type PriceSource interface {
Lookup(context.Context, string) (Price, error)
}
type Catalog map[string]int
func (c Catalog) Lookup(ctx context.Context, item string) (Price, error) {
if err := ctx.Err(); err != nil {
return Price{}, err
}
cents, ok := c[item]
if !ok {
return Price{}, fmt.Errorf("%w: %q", ErrUnknownItem, item)
}
return Price{Item: item, Cents: cents}, nil
}
type Service struct{ source PriceSource }
func NewService(source PriceSource) Service { return Service{source: source} }
func (s Service) Quote(ctx context.Context, item string) (Price, error) {
var price, shipping Price
_, err := RunTwo(ctx,
func(ctx context.Context) (string, error) {
var err error
price, err = s.source.Lookup(ctx, item)
return "", err
},
func(ctx context.Context) (string, error) {
var err error
shipping, err = s.source.Lookup(ctx, "shipping")
return "", err
})
if err != nil {
return Price{}, err
}
return Price{Item: item, Cents: price.Cents + shipping.Cents}, nil
}两个闭包分别查询商品与运费，各写一个局部变量。RunTwo 收到完成消息后，调用方才读取这些变量并计算总额。每个变量只有一个写入者，结果发送与接收建立同步关系。
Task 的字符串返回值用于通用执行例子；报价层把金额保留在 Price 中，并使用同一执行流程收取错误。
接入真实 Source 前，要确认它能被并发调用。本例 Catalog 创建后只读。若实现内部还有共享缓存，应在实现中安排同步。
看测试： 两个任务各发出“已开始”信号，再等待测试放行。测试先确认两项都启动，再允许返回。这个安排能验证并发关系，运行快慢不影响判断。
06：批量报价需要并发上限
如果商品列表有几千项，逐项启动 goroutine 会把压力传给数据源。这一阶段使用固定数量的 worker，任务下标通过 jobs channel 分发。
learning/go-patterns/stages/06-bounded/main.go · L1–L84 · 打开完整源码
// Stage 06: run a batch with a fixed number of workers.
package main
import (
"context"
"errors"
"fmt"
"sync"
)
// Task is one batch item. This stage focuses on limiting active work; the
// context-aware cancellation protocol is introduced in stage 07.
type Task func() (string, error)
type indexedTask struct {
index int
task Task
}
type indexedResult struct {
index int
value string
err error
}
// RunBounded runs every task with at most limit tasks active at once. It
// preserves input order and waits for the whole batch before returning. If
// several tasks fail, the error belonging to the lowest input index is
// returned so the result is deterministic.
func RunBounded(tasks []Task, limit int) ([]string, error) {
if limit < 1 {
return nil, errors.New("limit must be positive")
}
for index, task := range tasks {
if task == nil {
return nil, fmt.Errorf("task %d is nil", index)
}
}
values := make([]string, len(tasks))
if len(tasks) == 0 {
return values, nil
}
workerCount := limit
if workerCount > len(tasks) {
workerCount = len(tasks)
}
jobs := make(chan indexedTask)
results := make(chan indexedResult, len(tasks))
var workers sync.WaitGroup
workers.Add(workerCount)
for worker := 0; worker < workerCount; worker++ {
go func() {
defer workers.Done()
for job := range jobs {
value, err := job.task()
results <- indexedResult{index: job.index, value: value, err: err}
}
}()
}
// Sending from this goroutine applies back-pressure: once all workers are
// busy, dispatch waits until one worker takes another job.
for index, task := range tasks {
jobs <- indexedTask{index: index, task: task}
}
close(jobs)
workers.Wait()
close(results)
errs := make([]error, len(tasks))
for result := range results {
values[result.index] = result.value
errs[result.index] = result.err
}
for _, err := range errs {
if err != nil {
return values, err
}
}
return values, nil
}读这一组语句时，先找到责任分工：
- 调用方创建 jobs，启动 worker，再发送下标。
- worker 从 jobs 取下标，把结果写回对应位置。
- 发送方结束后关闭 jobs。
- worker 的 range 结束，defer 调用 Done。
- 调用方 Wait 返回后，才检查错误并读取整批结果。
不同 worker 写不同的 slice 元素，每个下标只分配一次。长度和底层存储在启动前已经固定，工作期间没有 append。Wait 把结果读取放在全部写入结束之后，因此这里可以省去额外的结果锁。
本阶段会把全部任务做完，再按输入下标报告错误。它适合希望收齐整批工作的场景。若结果接口要保留每项错误，可以进一步返回按项组织的结果结构。
并发上限放在哪一层
learning/go-patterns/stages/06-bounded/quote.go · L1–L53 · 打开完整源码
package main
import (
"context"
"errors"
"fmt"
)
var ErrUnknownItem = errors.New("unknown item")
type Price struct {
Item string
Cents int
}
type PriceSource interface {
Lookup(context.Context, string) (Price, error)
}
type Catalog map[string]int
func (c Catalog) Lookup(ctx context.Context, item string) (Price, error) {
if err := ctx.Err(); err != nil {
return Price{}, err
}
cents, ok := c[item]
if !ok {
return Price{}, fmt.Errorf("%w: %q", ErrUnknownItem, item)
}
return Price{Item: item, Cents: cents}, nil
}
type Service struct{ source PriceSource }
func NewService(source PriceSource) Service { return Service{source: source} }
func (s Service) Quote(ctx context.Context, item string) (Price, error) {
price, err := s.source.Lookup(ctx, item)
if err != nil {
return Price{}, err
}
shipping, err := s.source.Lookup(ctx, "shipping")
if err != nil {
return Price{}, err
}
return Price{Item: item, Cents: price.Cents + shipping.Cents}, nil
}
func (s Service) QuoteMany(ctx context.Context, items []string, limit int) ([]Price, error) {
quotes := make([]Price, len(items))
tasks := make([]Task, len(items))
for i, item := range items {
tasks[i] = func() (string, error) { quote, err := s.Quote(ctx, item); quotes[i] = quote; return "", err }
}
_, err := RunBounded(tasks, limit)
return quotes, err
}这里把并发安排在“整份报价”这一层，每个 worker 内依次查询商品和运费。因此最多同时有 limit 份报价进行，底层 Lookup 的同时执行数也受这个数量限制。
这个选择避免在每份报价里再开两个 goroutine，使底层请求数变成上限的两倍。若业务确实要保留单份报价内部的并发，就需要重新说明上限约束的是报价数还是外部请求数。
ctx 由闭包捕获，再传给 Quote。取消后，Source 能尽快返回错误；本阶段的调度器仍会走完整批任务。下一阶段会让调度本身也观察取消。
07：首个错误触发取消，最后统一等待
这一阶段保留 worker 数量上限，再加入 child context 和第一次失败的记录。
learning/go-patterns/stages/07-cancel-join/main.go · L1–L65 · 打开完整源码
// Stage 07 keeps the concurrency limit and adds cancellation plus joining.
package main
import (
"context"
"errors"
"fmt"
"sync"
)
// Long-running tasks must observe ctx or pass it to a cancellable API.
type Task func(context.Context) (string, error)
// RunFailFast returns ordered results after every worker has stopped.
// Unstarted tasks retain empty results. The first worker to record an error
// cancels the group; simultaneous failures have no fixed winner.
func RunFailFast(parent context.Context, tasks []Task, limit int) ([]string, error) {
if limit < 1 {
return nil, errors.New("limit must be positive")
}
for i, task := range tasks {
if task == nil {
return nil, fmt.Errorf("task %d is nil", i)
}
}
values := make([]string, len(tasks))
ctx, cancel := context.WithCancel(parent)
defer cancel()
jobs := make(chan int)
var workers sync.WaitGroup
var first sync.Once
var firstErr error
for range min(limit, len(tasks)) {
workers.Add(1)
go func() {
defer workers.Done()
for index := range jobs {
if ctx.Err() != nil {
return
}
value, err := tasks[index](ctx)
values[index] = value
if err != nil {
first.Do(func() { firstErr = err; cancel() })
return
}
}
}()
}
dispatch:
for index := range tasks {
select {
case jobs <- index:
case <-ctx.Done():
break dispatch
}
}
close(jobs)
workers.Wait()
if firstErr != nil {
return values, firstErr
}
return values, parent.Err()
}从失败那一刻追一遍：
- 一个 worker 得到错误，通过 Once 记录它。
- 这个 worker 调用 cancel，其他任务收到同一 child context 的取消信号。
- 发送下标的一侧也在观察取消，随后停止分发并关闭 jobs。
- 已经运行的任务协作退出，worker 执行 Done。
- Wait 返回后，函数才交出结果和错误。
如果多个任务同时失败，最先记录的错误取决于执行时序。调用方可以依赖“返回一个实际发生的失败”，需要稳定顺序时则应选择另一种聚合策略。
取消与发送都就绪时，select 可能仍完成一次发送。worker 在执行任务前再次检查 context，减少取消后的额外启动；这个检查与任务启动之间仍可能发生取消。真正耗时的操作也需要接收和使用 ctx。
业务层怎样使用这组能力
learning/go-patterns/stages/07-cancel-join/quote.go · L1–L57 · 打开完整源码
package main
import (
"context"
"errors"
"fmt"
)
var ErrUnknownItem = errors.New("unknown item")
type Price struct {
Item string
Cents int
}
type PriceSource interface {
Lookup(context.Context, string) (Price, error)
}
type Catalog map[string]int
func (c Catalog) Lookup(ctx context.Context, item string) (Price, error) {
if err := ctx.Err(); err != nil {
return Price{}, err
}
cents, ok := c[item]
if !ok {
return Price{}, fmt.Errorf("%w: %q", ErrUnknownItem, item)
}
return Price{Item: item, Cents: cents}, nil
}
type Service struct{ source PriceSource }
func NewService(source PriceSource) Service { return Service{source: source} }
func (s Service) Quote(ctx context.Context, item string) (Price, error) {
price, err := s.source.Lookup(ctx, item)
if err != nil {
return Price{}, err
}
shipping, err := s.source.Lookup(ctx, "shipping")
if err != nil {
return Price{}, err
}
return Price{Item: item, Cents: price.Cents + shipping.Cents}, nil
}
func (s Service) QuoteMany(ctx context.Context, items []string, limit int) ([]Price, error) {
quotes := make([]Price, len(items))
tasks := make([]Task, len(items))
for i, item := range items {
tasks[i] = func(taskCtx context.Context) (string, error) {
quote, err := s.Quote(taskCtx, item)
quotes[i] = quote
return "", err
}
}
_, err := RunFailFast(ctx, tasks, limit)
return quotes, err
}关键在任务闭包的参数：它接收执行器传入的 taskCtx，再用它调用 Quote。如果这里继续使用外面的原始 ctx，执行器发出的组内取消就传不到数据源。
未启动的任务保留零值结果。调用方要先检查错误，再决定是否使用已经完成的部分；公开 API 也可以为每项结果加上明确状态。
测试应该证明什么
这三个阶段的测试分别检查：
- 两项独立任务能在任一项返回前都开始。
- 同时运行的任务数没有超过指定上限。
- 结果按输入顺序保存。
- 一项失败后，其他协作任务收到了取消。
- 函数返回时，这些任务已经退出。
测试用 channel 控制先后关系，超时只负责在出现死锁时结束测试。race 检查可以发现测试覆盖到的并发访问问题，读代码时仍要解释每份状态的同步方式。
cd learning/go-patterns
go test -race ./stages/05-parallel ./stages/06-bounded ./stages/07-cancel-join
go run ./stages/07-cancel-join阶段 07 的 demo 会输出数据源失败，然后确认 worker 已停止。报价的成功路径由每个阶段的 quote_test.go 验证：book 总额 2600 分，pen 总额 800 分。
实际项目中的选择
任务量固定且很少时，可以直接启动几项工作并收齐。批量任务需要先确定并发上限的含义。失败后的策略也应由需求决定：继续收齐、停止分发，或者取消整组。
Go 项目也经常使用 golang.org/x/sync/errgroup 简化这类代码。理解了 context、错误记录和等待的关系，再读它的 API 会更容易。这里保留标准库实现，方便直接看清各部分怎样配合。
回到 Spine，可以继续观察工具函数、每轮状态和 Future 怎样组合在真实运行时里。