ShowDocs
搜索文章Ctrl K

外观

← 返回Go列表
开发实践

Go 高级并发实践：有界执行、背压、失败取消与保序 ​

前置：context取消、内存模型、GMP、GC预算。本项目把高级机制连成一个完整实验：把整数任务分发给固定数量的工作者，保持输出顺序，任一任务失败后取消其他工作，并在返回前等待退出。Go1.27.2，只用标准库，Windows/Linux 的执行属于历史验证，本轮未执行示例和测试；macOS 仅保留平台说明，本项目不执行平台运行验证。

contract：有界并发、背压与失败语义 ​

接口为 Run(ctx context.Context, workers int, input []int, work func(context.Context,int)(int,error)) ([]int,error)。ctx必须非nil；workers范围1–64；work不能nil。input在调用期间由调用者保持不变；回调可并发运行，不得无保护地写共享状态。成功返回与输入等长且按输入顺序排列的新结果；空输入返回长度0结果。任一错误或取消返回nil结果，已完成的部分结果不对外发布。

这里的有界指活跃工作者与待派发队列：最多 min(workers,len(input)) 个工作 G，jobs 是无缓冲 channel，生产者等接收形成背压，不为每个输入创建一个 G。input 和 output 仍是 O(n) 内存，不适合无限流；流式场景需增量输入、有限结果缓存和不同接口，不能把本例叫总内存固定。

首次记录到的回调错误触发子context取消；多个任务同时失败时哪一个先记录不固定。父context取消也会传播。取消是协作协议，已进入回调的任务必须自己响应 ctx；发送与取消同时就绪时可能派出额外任务，检查也不能与回调启动构成不可分割动作，因此不承诺“取消后绝不再执行任何一条指令”。Run承诺返回前所有工作者退出，不承诺能强杀忽略ctx的回调。业务副作用需幂等/事务，取消不回滚已写出的数据。

有界执行器完整实现：模块与退出责任 ​

仓库路径 examples/go/system-lab。独立练习创建同名文件和子目录，保留模块名以匹配 cmd/lab 的 import；没有外部依赖，不需要下载数据库或框架。go.mod如下；格式化实验的另两个文件在性能实验，按需一起保存。

go.mod
text
module showdocs.example/system-lab

go 1.27.0
worker.go（完整实现）
go
package systemlab

import (
	"context"
	"errors"
	"sync"
)

var ErrWorkers = errors.New("workers must be between 1 and 64")
var ErrWork = errors.New("work is required")

// Run uses at most workers callbacks concurrently and returns results in input
// order. Work must honor cancellation and must not mutate shared input state.
// On error Run discards partial results and joins all workers before returning.
func Run(ctx context.Context, workers int, input []int, work func(context.Context, int) (int, error)) ([]int, error) {
	if workers < 1 || workers > 64 {
		return nil, ErrWorkers
	}
	if work == nil {
		return nil, ErrWork
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	output := make([]int, len(input))
	jobs := make(chan int) // no unbounded queue or goroutine per input
	var wg sync.WaitGroup
	var once sync.Once
	var firstError error
	for range min(workers, len(input)) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case i, ok := <-jobs:
					if !ok {
						return
					}
					if ctx.Err() != nil {
						return
					}
					value, err := work(ctx, input[i])
					if err != nil {
						once.Do(func() { firstError = err; cancel() })
						return
					}
					output[i] = value
				}
			}
		}()
	}
dispatch:
	for i := range input {
		select {
		case <-ctx.Done():
			break dispatch
		case jobs <- i:
		}
	}
	close(jobs) // only this producer closes jobs
	wg.Wait()
	if firstError != nil {
		return nil, firstError
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return output, nil
}
cmd/lab/main.go（完整运行与追踪）
go
package main

import (
	"context"
	"fmt"
	"os"
	"runtime/trace"
	"showdocs.example/system-lab"
)

func run() error {
	file, err := os.Create("trace.out")
	if err != nil {
		return err
	}
	defer file.Close()
	if err = trace.Start(file); err != nil {
		return err
	}
	defer trace.Stop()
	ctx, task := trace.NewTask(context.Background(), "bounded-batch")
	defer task.End()
	output, err := systemlab.Run(ctx, 2, []int{1, 2, 3, 4}, func(ctx context.Context, n int) (int, error) {
		region := trace.StartRegion(ctx, "square")
		defer region.End()
		return n * n, nil
	})
	if err != nil {
		return err
	}
	fmt.Println(output)
	return nil
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

每个工作者只写output独占下标，输入下标由唯一生产者分发，不发生相同元素的并发写；不同元素可以独立写。firstError由sync.Once单次写入；主 G 在wg.Wait返回后读它，避免和写者并发。生产者是jobs唯一关闭者，工作者只接收；Wait保证所有写入完成后才发布output。不要把close挪到任一工作者，否则其他工作者/生产者可能发送到已关闭channel。

循环中的select让工作者能在等待任务时退出，dispatch的select让生产者能在无接收者时退出。子context在每次函数退出都cancel，避免资源残留。用户回调panic不在这里恢复；panic会终止进程，而不是伪装成正常业务错误。业务层确有隔离要求时需设计错误契约、清理和恢复边界，不能加一个空recover吞掉问题。

任务取消后卡住：测试与诊断 ​

进入模块目录执行，PowerShell/Bash 通用：

sh
go version
go test -count=10 ./...
go vet ./...
go run ./cmd/lab
go test -run TestRun -trace=test-trace.out .
go tool trace -d=parsed test-trace.out

预期测试通过，运行输出 [1 4 9 16]，工作者完成后进程正常退出。运行会覆盖实验目录trace.out；测试追踪写test-trace.out。编译/测试不执行刻意错误的cmd/race程序，只有内存模型反例的显式命令会运行它。测试与主程序使用同一个worker.go，不存在另写一套“看似正确”的伪实现。

worker_test.go（完整行为测试）

完整文件：worker_test.go 完整源码。保留原始内容，使用前请核对本文的版本与环境说明。

测试通过channel屏障确认工作已开始，再解除/取消，不靠Sleep猜测时序。5秒计时仅是失效时的测试看门狗，不参与正确性协议。测试检查最多两个活跃回调、返回时活跃数为零、错误被保留、已取消输入不启动工作、输出保序及输入不变；测试不是对任意恶意回调的形式证明。支持竞态检测的环境再运行 go test -race -count=10 ./...，实际平台结果与工具链限制如实记录。

排查卡住时先检查所有阻塞点：生产者是否能观察取消，工作者是否仍在不可取消回调，谁负责关闭，谁负责等待；用trace/goroutine栈确认等待位置。不要为“消除卡住”直接删Wait，那会让函数返回后仍有后台写入甚至操作已释放资源。更多诊断见性能实验。

exercise：阶段综合练习与取舍 ​

  1. 扩充正常任务到不同耗时但保序的计算，解释执行完成顺序与结果排列的差别；用channel明确控制先后，不用固定Sleep做测试。
  2. 在单个回调中加入可取消的阻塞步骤，先触发开始屏障再取消；确认返回的是context错误，且活跃数归零。
  3. 若输入是几百万项，说明当前内存下界；设计流式版本时写清结果背压、错误后的已发结果和排序缓冲上限，不必现在实现通用框架。
  4. 用相同任务负载分别设workers=1、2、4，记录正确性、吞吐及资源。说明CPU密集和下游受限任务为何可能需要不同并发数。
参考解法

输出下标由输入索引决定，后完成的前项仍放在较小下标；用屏障让后项先完成再释放前项，最后断言完整结果。取消测试已在worker_test中展示：回调先通知started，再等ctx.Done，Run返回后active必须为0。当前输入/输出各与n成正比；有序流若前项很慢而无限接收后项会造成排序缓存膨胀，应限制未确认窗口或允许乱序并显式携带ID。提高workers可能增加队列/连接占用和调度开销；先确认下游容量和瓶颈，不能以CPU核心数作为所有场景的固定答案。

依据：context取消约定、WaitGroup与Once、Go流水线与取消、内存模型。核对日期2026-10-17。该教学模块不提供生产任务持久化、重试、鉴权或远程调度。

继续阅读