跳到主要内容

Go 并发实战：从零实现一个高并发安全的任务调度器

· 阅读需 25 分钟
ahKevinXy
ahKevinXy
作者

批量生成报表、发送通知、处理图片时，一个常见写法是：每收到一个任务，就启动一个 goroutine。

任务数量少时，这种写法很方便。请求突然增多后，问题却往往出现在数据库连接池、第三方接口和内存上：执行中的任务越来越多，等待资源的 goroutine 也持续积累，服务难以控制负载，更难判断什么时候可以安全退出。

这一篇用 Go 标准库实现一个单进程、即时执行的任务调度器：固定数量的 worker 执行任务，有界队列吸收短时积压，队列满时明确拒绝，关闭时停止接收并等待已接收任务结束。

重点是把提交、执行、取消和停机的并发语义说清楚。这里不实现 Cron、定时触发或分布式任务分配，也不提供进程崩溃后的任务恢复。

一、先确定调度器承诺什么​

假设一个报表服务最多允许 8 个报表同时生成，并允许额外排队 256 个任务。我们的调度器需要满足：

场景约定
多个 goroutine 同时提交调度器内部状态安全，任务成功入队后不会被正常关闭流程丢弃
执行资源达到上限同时执行的任务不超过 worker 数量
队列已经满了立即返回 ErrQueueFull，不等待队列空位
任务执行失败通过结果 channel 返回任务错误
任务发生普通 panic在对应 worker 内恢复，返回 PanicError，worker 继续工作
排队任务的 context 已取消worker 取到任务后跳过执行，返回 context 错误
开始停机拒绝新任务，排空队列，等待 worker 退出
停机等待超时返回等待错误；已接收任务仍按原来的 context 继续处理

这里的“并发安全”有明确边界：它保护调度器自己的状态，不会自动保护任务闭包访问的业务 map、slice 或指针。成功接收也不等于执行成功，更不等于任务已经持久化。

队列中的任务按入队顺序被取出；多个 worker 何时开始执行、什么时候完成，不保证按照提交顺序排列。

二、用固定 worker 和有界队列组织任务​

整体结构可以画成这样：

我们只提供三个公开入口：

New(workers, queueSize int) (*Scheduler, error)
TrySubmit(ctx context.Context, task Task) (<-chan error, error)
Shutdown(ctx context.Context) error

TrySubmit 返回两层结果：

  • 方法返回的 error：任务是否被接收，例如队列已满、调度器已关闭。
  • 结果 channel 中的 error：被接收的任务最终是否执行成功。

例如：

result, err := s.TrySubmit(taskCtx, generateReport)
if err != nil {
    // 没有接收任务，可由上层返回繁忙、降级或安排重试。
    return err
}

err = <-result // nil 表示这次任务正常返回。

TrySubmit 不等待队列空位，但仍可能等待提交状态锁。它不是无锁算法，也不保证在任意竞争程度下都能立即返回。

三、完整实现​

新建一个独立示例目录：

task-scheduler/
├── go.mod
├── scheduler.go
├── scheduler_test.go
└── cmd/demo/main.go

go.mod 使用 Go 1.21 语言版本，示例没有第三方依赖；本次编译和竞态测试使用 Go 1.27.3。语言兼容版本不等于部署工具链版本，实际运行应使用仍受维护的 Go 版本。

go.mod
module example.com/taskscheduler

go 1.21

下面是完整的 scheduler.go。后面的分析都对应这份实现。

scheduler.go
// Package scheduler implements a bounded, in-process task scheduler.
package scheduler

import (
	"context"
	"errors"
	"fmt"
	"runtime/debug"
	"sync"
)

var (
	ErrClosed     = errors.New("scheduler is closed")
	ErrQueueFull  = errors.New("scheduler queue is full")
	ErrNilContext = errors.New("context must not be nil")
	ErrNilTask    = errors.New("task must not be nil")
)

// Task owns any synchronization needed by the data it captures. Cancellation is
// cooperative: a task must inspect ctx or pass it to context-aware operations.
// Tasks must return; runtime.Goexit, os.Exit and child goroutine panics are not
// isolated by this scheduler.
type Task func(context.Context) error

// PanicError records a panic raised by the task on its worker goroutine.
type PanicError struct {
	Value any
	Stack []byte
}

func (e *PanicError) Error() string { return fmt.Sprintf("task panic: %v", e.Value) }

// Scheduler runs at most workers tasks simultaneously and queues at most
// queueSize tasks. It must not be copied after first use.
type Scheduler struct {
	mu     sync.Mutex
	closed bool
	jobs   chan job
	wg     sync.WaitGroup
	done   chan struct{}
}

type job struct {
	ctx    context.Context
	task   Task
	result chan error
}

// New starts a fixed number of workers. Both arguments must be positive.
func New(workers, queueSize int) (*Scheduler, error) {
	if workers <= 0 || queueSize <= 0 {
		return nil, fmt.Errorf("workers and queueSize must be positive")
	}
	s := &Scheduler{
		jobs: make(chan job, queueSize),
		done: make(chan struct{}),
	}
	s.wg.Add(workers)
	for i := 0; i < workers; i++ {
		go s.worker()
	}
	go func() {
		s.wg.Wait()
		close(s.done)
	}()
	return s, nil
}

// TrySubmit never waits for queue space: it returns ErrQueueFull when the queue
// is full. It may briefly wait for the admission mutex. A nil admission error
// means accepted, not completed. The returned channel delivers exactly one task
// result and is then closed; callers may ignore it without blocking a worker.
// Cancellation can race with acceptance; an accepted canceled job is completed
// with ctx.Err() if cancellation is observed before the task starts.
func (s *Scheduler) TrySubmit(ctx context.Context, task Task) (<-chan error, error) {
	if ctx == nil {
		return nil, ErrNilContext
	}
	if task == nil {
		return nil, ErrNilTask
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return nil, ErrClosed
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	result := make(chan error, 1)
	select {
	case s.jobs <- job{ctx: ctx, task: task, result: result}:
		return result, nil
	default:
		return nil, ErrQueueFull
	}
}

func (s *Scheduler) worker() {
	defer s.wg.Done()
	for j := range s.jobs {
		j.result <- run(j.ctx, j.task)
		close(j.result)
	}
}

func run(ctx context.Context, task Task) (err error) {
	if err := ctx.Err(); err != nil {
		return err
	}
	defer func() {
		if value := recover(); value != nil {
			err = &PanicError{Value: value, Stack: debug.Stack()}
		}
	}()
	return task(ctx)
}

// Shutdown closes admission, drains accepted jobs and waits for all workers.
// Concurrent calls are safe. ctx limits only this call's wait; a timeout neither
// cancels accepted tasks nor reopens admission. A later call can wait again.
// Do not call Shutdown synchronously from a task on this scheduler: that task
// would be waiting for its own worker to finish.
func (s *Scheduler) Shutdown(ctx context.Context) error {
	if ctx == nil {
		return ErrNilContext
	}
	s.mu.Lock()
	if !s.closed {
		s.closed = true
		close(s.jobs)
	}
	s.mu.Unlock()

	// An already completed shutdown is successful even with an expired context.
	select {
	case <-s.done:
		return nil
	default:
	}
	select {
	case <-s.done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

使用时始终通过 New 创建实例，并传递 *Scheduler；不要复制它，也不要直接使用零值实例。

1. 提交和关闭使用同一把锁​

并发调度器最容易出现的问题之一，是一个 goroutine 正在发送任务，另一个 goroutine 已经关闭队列。向已关闭的 channel 发送数据会触发 panic。Go 语言规范 对这一行为有明确规定。

单独写一个原子布尔值不能解决问题：

if !closed.Load() {
    jobs <- task // 检查之后、发送之前，其他 goroutine 仍可能关闭 jobs。
}

本实现让以下操作进入同一个临界区：

  1. TrySubmit 检查 closed，然后尝试发送。
  2. Shutdown 把 closed 设为 true，然后关闭队列。

成功发送是任务被接收的时刻。关闭流程持有锁时，不会有提交方同时向队列发送；提交方持有锁时，关闭流程也不能提前关闭队列。

关键还在于发送使用了带 default 的 select。如果持锁执行阻塞发送，队列满时提交方就会一直占着锁，关闭流程也无法取得锁。这里的临界区只做状态判断和一次入队尝试，不执行任务、不等待任务结果，也不等待队列空位。

2. worker 数量固定，任务在锁外执行​

New 在启动 worker 前一次性调用 wg.Add(workers)，WaitGroup 统计的是 worker 生命周期。它没有对每次提交动态执行 Add，因此关闭流程不需要处理“开始 Wait 时还有新任务正在 Add”的协调问题。

每个 worker 使用 for j := range s.jobs 消费队列。关闭一个仍有缓冲数据的 channel 后，已有数据仍然可以被取出，worker 会处理完这些任务再退出。

任务代码运行在提交状态锁之外。因此，一个任务访问网络或数据库时，不会一直占着用于提交和关闭的那把锁。

3. 结果 channel 为什么容量为 1​

每个任务只有一个最终结果，所以用容量为 1 的 channel 保存它。

如果使用无缓冲 channel，而调用方不读取结果，worker 就会阻塞在发送结果的位置：任务明明执行结束了，worker 却不能处理下一项任务，Shutdown 也可能一直等不到它退出。

有了一个缓冲位，即使调用方放弃读取，worker 仍能交付结果并继续工作。发送结果后再关闭 channel，也明确表示这个任务不会再产生第二个结果。

注意，这个 channel 是一次消费的结果，不是可以广播给多个等待者的 Future。重复读取已关闭的 channel 会得到零值；如果需要区分，应检查 err, ok := <-result。多个读取方不能依赖它们都拿到真实的任务错误。

四、取消和异常处理不能只看方法名字​

排队超时与执行超时​

每个任务携带自己的 context。worker 在调用任务前检查 ctx.Err()，已经取消的排队任务会直接产生错误结果。

这并不会立即从队列中删除任务：任务仍占据一个队列位置，直到 worker 把它取出。若业务经常批量取消大量积压任务，需要进一步设计可删除的队列或专门的调度循环。

取消和提交也可能同时发生。TrySubmit 检查 context 后，取消信号仍可能在入队前到达；worker 检查之后，context 也可能马上取消。因此，执行中的任务必须自己处理取消，或者把 context 传给支持它的数据库、HTTP 等操作。context 是协作式取消机制，不会强制终止任意 goroutine。context 文档 说明了取消信号的传递方式。

任务中的等待可以这样写：

func generateReport(ctx context.Context) error {
    timer := time.NewTimer(100 * time.Millisecond)
    defer timer.Stop()

    select {
    case <-timer.C:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

如果在提交之前创建超时 context，预算包含排队时间和执行时间。如果还需要独立的“开始执行后最多运行多久”，可以在任务函数内再派生执行超时，但它仍受父 context 的截止时间约束。

对于响应结束后仍需运行的后台任务，不要直接依赖即将结束的 HTTP 请求 context。应由服务生命周期管理任务 context、截止时间和必要的业务数据，避免后台任务刚入队就随请求一起被取消。

panic 恢复要放在每次任务调用边界​

recover 放在 run 中，而不是只放在整个 worker 的最外层。这样一次任务发生普通 panic 时，run 会返回 PanicError，worker 随后仍能继续下一轮循环。

调用方可以区分任务错误和 panic：

err := <-result
var panicErr *scheduler.PanicError
if errors.As(err, &panicErr) {
    log.Printf("task panicked: %v\n%s", panicErr.Value, panicErr.Stack)
} else if err != nil {
    log.Printf("task failed: %v", err)
}

恢复范围仅限于执行任务的那个 goroutine。任务自己启动的子 goroutine 若发生未恢复的 panic，不会被这里捕获；runtime.Goexit 会结束 worker，也不是普通 panic。任务必须遵守返回约定，不能调用 runtime.Goexit、os.Exit，也不应在任务中使用会退出测试 goroutine 的 t.Fatal。

五、优雅停机是一个不可逆的状态转换​

关闭流程分成两步：先关闭入口，再等待执行结束。

接收任务 → 停止接收并排空队列 → 所有 worker 退出

第一次 Shutdown 在锁内设置关闭状态并关闭 jobs。后续调用只会等待同一个 done channel，因此多个关闭调用不会重复执行 close(jobs)。

等待发生在解锁之后。worker 不需要取得提交状态锁就能处理剩余任务并退出，所以关闭调用不会用这把锁阻挡正常排空。

Shutdown 的 context 只限制这次调用等待多久。即使传入的 context 已经取消，关闭入口的动作仍会先执行；超时也不会重新开放入口，更不会自动取消已接收任务。稍后可以用新的 context 再调用一次 Shutdown 继续等待。

还要避开两种容易忽略的自我等待：

  • 不要在一个任务内部，同步调用它所在调度器的 Shutdown(context.Background())；关闭流程会等待这个任务自己结束。
  • 不要让所有 worker 都提交子任务到同一个池，再同步等待子任务结果；子任务可能没有空闲 worker 可用。

停机应由服务入口或生命周期管理器负责。进程退出前是否能够完成排空，最终还取决于任务是否会返回、是否正确响应自己的 context。

六、跑通一个完整示例​

下面的入口启动 4 个 worker，允许 32 个任务排队，提交 20 个模拟任务，再关闭调度器并读取结果。

cmd/demo/main.go
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"sync/atomic"
	"time"

	scheduler "example.com/taskscheduler"
)

func main() {
	s, err := scheduler.New(4, 32)
	if err != nil {
		log.Fatal(err)
	}
	var completed atomic.Int64
	var results []<-chan error
	for i := 0; i < 20; i++ {
		id := i // Explicit capture also works with the module's Go 1.21 semantics.
		result, err := s.TrySubmit(context.Background(), func(ctx context.Context) error {
			select {
			case <-time.After(5 * time.Millisecond):
				completed.Add(1)
				return nil
			case <-ctx.Done():
				return fmt.Errorf("task %d: %w", id, ctx.Err())
			}
		})
		if errors.Is(err, scheduler.ErrQueueFull) {
			log.Printf("task %d rejected: queue full", id)
			continue
		}
		if err != nil {
			log.Fatal(err)
		}
		results = append(results, result)
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := s.Shutdown(ctx); err != nil {
		log.Fatal(err)
	}
	for _, result := range results {
		if err := <-result; err != nil {
			log.Printf("task failed: %v", err)
		}
	}
	fmt.Printf("accepted=%d completed=%d\n", len(results), completed.Load())
}

在示例模块目录执行：

cd task-scheduler
go run ./cmd/demo

正常输出为：

accepted=20 completed=20

这里显式写了 id := i，让闭包捕获当前循环值。模块声明的是 Go 1.21 语言版本，这样写也不依赖 Go 1.22 引入的新循环变量语义。

这个例子先等待关闭、再读取结果，能够成立的一个前提正是：每个结果 channel 都有一个缓冲位。实际业务也可以在自己的结果处理流程中逐项消费，但不要为每次尝试提交都无上限地创建等待 goroutine。

七、用测试验证并发边界​

仅仅跑通一次示例，不足以验证提交和关闭交错时的行为。下面给出完整的 scheduler_test.go，包含边界测试和并行基准；将它保存到前面的模块目录即可运行。测试使用 channel 协调执行时机，而不是靠随意增加 time.Sleep 来猜调度顺序。

重点检查这些情形：

测试场景必须验证的行为
worker 被可控任务占满，再填满队列下一次提交返回 ErrQueueFull
多个提交方与关闭操作同时发生每个成功接收的任务都有一个结果，拒绝项不混入完成计数
大量任务交错执行记录同时执行数量，峰值不超过 worker 数量
排队期间取消 context对应任务体没有执行，结果是取消错误
任务返回错误或发生 panic调用方得到正确错误，后续任务仍能执行
关闭等待超时，再次调用关闭第一次只结束等待，释放任务后第二次可以正常完成
调用方不读取结果worker 仍能继续消费任务，关闭可以完成
展开完整测试与基准代码（scheduler_test.go）
scheduler_test.go
package scheduler

import (
	"context"
	"errors"
	"runtime"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

const testTimeout = 5 * time.Second

func newScheduler(t *testing.T, workers, queueSize int) *Scheduler {
	t.Helper()
	s, err := New(workers, queueSize)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
		defer cancel()
		if err := s.Shutdown(ctx); err != nil {
			t.Errorf("cleanup Shutdown: %v", err)
		}
	})
	return s
}

func waitSignal(t *testing.T, signal <-chan struct{}) {
	t.Helper()
	select {
	case <-signal:
	case <-time.After(testTimeout):
		t.Fatal("timed out waiting for signal")
	}
}

func getResult(t *testing.T, result <-chan error) error {
	t.Helper()
	var err error
	select {
	case value, ok := <-result:
		if !ok {
			t.Fatal("result closed without a result")
		}
		err = value
	case <-time.After(testTimeout):
		t.Fatal("timed out waiting for result")
	}
	select {
	case _, ok := <-result:
		if ok {
			t.Fatal("result contains more than one value")
		}
	case <-time.After(testTimeout):
		t.Fatal("result channel was not closed")
	}
	return err
}

func mustSubmit(t *testing.T, s *Scheduler, ctx context.Context, task Task) <-chan error {
	t.Helper()
	result, err := s.TrySubmit(ctx, task)
	if err != nil {
		t.Fatal(err)
	}
	return result
}

func TestInvalidArguments(t *testing.T) {
	for _, args := range [][2]int{{0, 1}, {-1, 1}, {1, 0}, {1, -1}} {
		if s, err := New(args[0], args[1]); err == nil || s != nil {
			t.Fatalf("New(%d, %d) = %v, %v", args[0], args[1], s, err)
		}
	}
	s := newScheduler(t, 1, 2)
	if result, err := s.TrySubmit(nil, func(context.Context) error { return nil }); !errors.Is(err, ErrNilContext) || result != nil {
		t.Fatalf("nil context: result=%v err=%v", result, err)
	}
	if result, err := s.TrySubmit(context.Background(), nil); !errors.Is(err, ErrNilTask) || result != nil {
		t.Fatalf("nil task: result=%v err=%v", result, err)
	}
	if err := s.Shutdown(nil); !errors.Is(err, ErrNilContext) {
		t.Fatalf("nil shutdown context: %v", err)
	}
	// Invalid Shutdown input must not close admission.
	if err := getResult(t, mustSubmit(t, s, context.Background(), func(context.Context) error { return nil })); err != nil {
		t.Fatal(err)
	}
}

func TestQueueFullAndDrain(t *testing.T) {
	s := newScheduler(t, 1, 1)
	started, release := make(chan struct{}), make(chan struct{})
	var releaseOnce sync.Once
	defer releaseOnce.Do(func() { close(release) })
	first := mustSubmit(t, s, context.Background(), func(context.Context) error {
		close(started)
		<-release
		return nil
	})
	waitSignal(t, started)
	var queuedRan atomic.Bool
	second := mustSubmit(t, s, context.Background(), func(context.Context) error {
		queuedRan.Store(true)
		return nil
	})
	if result, err := s.TrySubmit(context.Background(), func(context.Context) error { return nil }); !errors.Is(err, ErrQueueFull) || result != nil {
		t.Fatalf("full queue: result=%v err=%v", result, err)
	}
	releaseOnce.Do(func() { close(release) })
	if err := s.Shutdown(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := getResult(t, first); err != nil {
		t.Fatal(err)
	}
	if err := getResult(t, second); err != nil || !queuedRan.Load() {
		t.Fatalf("queued job was not drained: result=%v ran=%v", err, queuedRan.Load())
	}
}

func TestMaximumRunningTasks(t *testing.T) {
	const workers, tasks = 3, 24
	s := newScheduler(t, workers, tasks)
	started, release := make(chan struct{}, tasks), make(chan struct{})
	var releaseOnce sync.Once
	defer releaseOnce.Do(func() { close(release) })
	var active, peak, completed atomic.Int64
	results := make([]<-chan error, 0, tasks)
	for i := 0; i < tasks; i++ {
		results = append(results, mustSubmit(t, s, context.Background(), func(context.Context) error {
			n := active.Add(1)
			defer active.Add(-1)
			for old := peak.Load(); n > old && !peak.CompareAndSwap(old, n); old = peak.Load() {
			}
			started <- struct{}{}
			<-release
			completed.Add(1)
			return nil
		}))
	}
	for i := 0; i < workers; i++ {
		waitSignal(t, started)
	}
	if got := active.Load(); got != workers {
		t.Fatalf("active=%d want=%d", got, workers)
	}
	releaseOnce.Do(func() { close(release) })
	if err := s.Shutdown(context.Background()); err != nil {
		t.Fatal(err)
	}
	for _, result := range results {
		if err := getResult(t, result); err != nil {
			t.Fatal(err)
		}
	}
	if peak.Load() > workers || completed.Load() != tasks || active.Load() != 0 {
		t.Fatalf("peak=%d completed=%d active=%d", peak.Load(), completed.Load(), active.Load())
	}
}

func TestCanceledBeforeSubmission(t *testing.T) {
	s := newScheduler(t, 1, 1)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	var ran atomic.Bool
	result, err := s.TrySubmit(ctx, func(context.Context) error { ran.Store(true); return nil })
	if !errors.Is(err, context.Canceled) || result != nil || ran.Load() {
		t.Fatalf("result=%v err=%v ran=%v", result, err, ran.Load())
	}
}

func TestDeadlineExpiresWhileQueued(t *testing.T) {
	s := newScheduler(t, 1, 1)
	started, release := make(chan struct{}), make(chan struct{})
	var releaseOnce sync.Once
	defer releaseOnce.Do(func() { close(release) })
	blocker := mustSubmit(t, s, context.Background(), func(context.Context) error {
		close(started)
		<-release
		return nil
	})
	waitSignal(t, started)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	var ran atomic.Bool
	queued := mustSubmit(t, s, ctx, func(context.Context) error { ran.Store(true); return nil })
	<-ctx.Done() // A real deadline, not a sleep used to guess worker scheduling.
	releaseOnce.Do(func() { close(release) })
	if err := getResult(t, queued); !errors.Is(err, context.DeadlineExceeded) || ran.Load() {
		t.Fatalf("expired job: err=%v ran=%v", err, ran.Load())
	}
	if err := getResult(t, blocker); err != nil {
		t.Fatal(err)
	}
}

func TestRunningTaskCooperatesWithCancellation(t *testing.T) {
	s := newScheduler(t, 1, 1)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	started := make(chan struct{})
	result := mustSubmit(t, s, ctx, func(ctx context.Context) error {
		close(started)
		<-ctx.Done()
		return ctx.Err()
	})
	waitSignal(t, started)
	cancel()
	if err := getResult(t, result); !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
}

func TestPanicIsReportedAndWorkerContinues(t *testing.T) {
	for _, value := range []any{"boom", nil} {
		value := value
		t.Run("panic", func(t *testing.T) {
			s := newScheduler(t, 1, 2)
			panicked := mustSubmit(t, s, context.Background(), func(context.Context) error { panic(value) })
			following := mustSubmit(t, s, context.Background(), func(context.Context) error { return nil })
			var panicErr *PanicError
			if err := getResult(t, panicked); !errors.As(err, &panicErr) || len(panicErr.Stack) == 0 {
				t.Fatalf("panic result=%v", err)
			}
			if err := getResult(t, following); err != nil {
				t.Fatalf("worker did not survive: %v", err)
			}
		})
	}
}

func TestTaskErrorIsPreserved(t *testing.T) {
	s := newScheduler(t, 1, 1)
	want := errors.New("business failure")
	result := mustSubmit(t, s, context.Background(), func(context.Context) error { return want })
	if err := getResult(t, result); !errors.Is(err, want) {
		t.Fatalf("got %v, want original error", err)
	}
}

func TestUnconsumedResultsDoNotBlockShutdown(t *testing.T) {
	s := newScheduler(t, 2, 32)
	var completed atomic.Int64
	for i := 0; i < 32; i++ {
		mustSubmit(t, s, context.Background(), func(context.Context) error { completed.Add(1); return nil })
	}
	ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
	defer cancel()
	if err := s.Shutdown(ctx); err != nil {
		t.Fatal(err)
	}
	if completed.Load() != 32 {
		t.Fatalf("completed=%d", completed.Load())
	}
}

func TestShutdownTimeoutDoesNotCancelTasks(t *testing.T) {
	s := newScheduler(t, 1, 1)
	started, release := make(chan struct{}), make(chan struct{})
	var releaseOnce sync.Once
	defer releaseOnce.Do(func() { close(release) })
	result := mustSubmit(t, s, context.Background(), func(context.Context) error {
		close(started)
		<-release
		return nil
	})
	waitSignal(t, started)
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	if err := s.Shutdown(ctx); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("Shutdown=%v", err)
	}
	if _, err := s.TrySubmit(context.Background(), func(context.Context) error { return nil }); !errors.Is(err, ErrClosed) {
		t.Fatalf("admission reopened: %v", err)
	}
	select {
	case err := <-result:
		t.Fatalf("Shutdown timeout incorrectly ended task: %v", err)
	default:
	}
	releaseOnce.Do(func() { close(release) })
	if err := s.Shutdown(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := getResult(t, result); err != nil {
		t.Fatal(err)
	}
	canceled, stop := context.WithCancel(context.Background())
	stop()
	if err := s.Shutdown(canceled); err != nil {
		t.Fatalf("already completed shutdown: %v", err)
	}
}

func TestConcurrentSubmitAndShutdown(t *testing.T) {
	const producers, attempts, closers = 64, 64, 16
	s := newScheduler(t, 8, 64)
	start := make(chan struct{})
	results := make(chan (<-chan error), producers*attempts+1)
	errorsSeen := make(chan error, producers*attempts)
	var accepted, executed atomic.Int64
	task := func(context.Context) error { executed.Add(1); return nil }
	results <- mustSubmit(t, s, context.Background(), task)
	accepted.Add(1)
	var producersWG, ready sync.WaitGroup
	producersWG.Add(producers)
	ready.Add(producers)
	for i := 0; i < producers; i++ {
		go func() {
			defer producersWG.Done()
			for j := 0; j < attempts; j++ {
				if j == 1 {
					ready.Done()
					<-start
				}
				result, err := s.TrySubmit(context.Background(), task)
				switch {
				case err == nil:
					accepted.Add(1)
					results <- result
				case errors.Is(err, ErrClosed), errors.Is(err, ErrQueueFull):
				default:
					errorsSeen <- err
				}
			}
		}()
	}
	ready.Wait() // Every producer has already attempted submission once.
	shutdownErrors := make(chan error, closers)
	for i := 0; i < closers; i++ {
		go func() {
			<-start
			ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
			defer cancel()
			shutdownErrors <- s.Shutdown(ctx)
		}()
	}
	close(start)
	producersWG.Wait()
	close(results)
	close(errorsSeen)
	for err := range errorsSeen {
		t.Errorf("unexpected admission error: %v", err)
	}
	for i := 0; i < closers; i++ {
		if err := <-shutdownErrors; err != nil {
			t.Errorf("concurrent Shutdown: %v", err)
		}
	}
	var delivered int64
	for result := range results {
		if err := getResult(t, result); err != nil {
			t.Errorf("accepted task: %v", err)
		}
		delivered++
	}
	if accepted.Load() != executed.Load() || accepted.Load() != delivered {
		t.Fatalf("accepted=%d executed=%d delivered=%d", accepted.Load(), executed.Load(), delivered)
	}
	if _, err := s.TrySubmit(context.Background(), task); !errors.Is(err, ErrClosed) {
		t.Fatalf("submission after Shutdown: %v", err)
	}
}

// This measures a no-op task including admission, its per-task result channel
// and completion delivery. Queue-full attempts retry, so ns/op includes retry
// cost. It is not a business workload or a claim of production throughput.
func BenchmarkTrySubmit(b *testing.B) {
	s, err := New(runtime.GOMAXPROCS(0), 1024)
	if err != nil {
		b.Fatal(err)
	}
	ctx := context.Background()
	noop := func(context.Context) error { return nil }
	b.ReportAllocs()
	b.ResetTimer()
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			for {
				result, err := s.TrySubmit(ctx, noop)
				if errors.Is(err, ErrQueueFull) {
					runtime.Gosched()
					continue
				}
				if err != nil {
					b.Error(err)
					return
				}
				if err := <-result; err != nil {
					b.Error(err)
				}
				break
			}
		}
	})
	b.StopTimer()
	if err := s.Shutdown(ctx); err != nil {
		b.Fatal(err)
	}
}

运行常规检查与竞态测试：

go vet ./...
go test ./...
go test -race -count=20 ./...

竞态检测器只能发现实际执行路径中触发的竞争，不能代替对关闭协议和状态约束的分析。因此需要把确定性的边界测试、重复并发测试与代码审查结合起来。Go 官方竞态检测文档 也明确指出了运行路径覆盖的限制。

Go 的内存模型定义了锁和 channel 通信的同步关系。本实现依靠同一把锁维护接收状态，再通过任务和结果 channel 交接数据，而不是依靠“看起来应该已经执行过”的时间顺序。Go 内存模型 是分析这些关系的基础。

但调度器无法替任务修复下面这样的业务竞争：

counts := map[string]int{}
// 多个任务同时执行这句，仍然需要业务自己的锁或其他并发设计。
counts["report"]++

向任务交接 slice、map、指针时，应明确谁拥有修改权。任务开始执行的时间甚至可能早于 TrySubmit 返回；不能把“提交方法返回了”当作可以随意改写共享数据的依据。

八、基准测试测什么，容量应该怎么选​

示例包含并行提交、执行并取回结果的基准测试。只有成功接收并完成的任务才算一次操作，不能把大量 ErrQueueFull 的快速返回当作业务吞吐量。

go test -run '^$' -bench . -benchmem -benchtime=1s -count=3 -cpu=1,4,8

ns/op、B/op 和 allocs/op 可以帮助观察一次调度往返的开销、每任务结果 channel 等分配成本，以及不同 CPU 并行度下的锁竞争。空任务基准反映的是调度开销，不是报表服务或网络请求的真实吞吐量；开启 -race 的耗时也不应拿来作为正常性能数据。

尤其要注意，RunParallel 的 ns/op 是整组基准的墙钟时间除以操作次数，不是每个任务的端到端延迟，更不能替代 P95、P99 排队与执行延迟统计。testing 文档 对并行基准的计时有专门说明。

容量规划时，先关注业务资源，再调 worker 数量：

  • CPU 密集任务可从接近 GOMAXPROCS 的并发数开始测试。
  • I/O 密集任务应结合数据库连接数、外部接口限流、内存和任务耗时测量，不要盲目扩大并发。
  • 队列用于承接短时积压；持续过载时，需要上游降速、拒绝或持久化转移，而不是无限扩容内存队列。

例如目标到达量是每秒 200 个任务，单任务平均执行时间为 40ms，那么仅从平均工作量估算，就需要约 200 × 0.04 = 8 个持续忙碌的执行位置。这只是粗略估算，不能保证尾延迟；还需给耗时波动、资源竞争和突发流量留出余量。

设 worker 数量为 W，队列容量为 C，在任务遵守执行约定的前提下，这个实现内部满足：

已接收且尚未完成的任务数量 ≤ W + C

这个约束不包含任务自己创建的 goroutine、调用方保留的已完成结果 channel，也不限制单个任务闭包捕获的数据大小。因此，“有界任务数量”不等于整个进程的内存使用有严格上限。

队列满时，上层可以拒绝请求、降级、采用有退避和次数上限的重试，或转交持久化队列。不要在业务代码里无限忙循环重试，也不要让“重试队列”变成另一个无界缓存。

九、继续扩展时，先补语义再加功能​

这个调度器适合进程内、允许通过明确错误处理拥塞的后台任务。继续扩展之前，先确定需要的保障：

需求需要补充的设计
失败后自动重试可重试错误分类、退避、次数上限、业务幂等
指定时间执行或周期任务定时触发层、时间管理，以及触发后队列满时的策略
任务有不同优先级带优先级的队列及防止低优先级任务长期饥饿的规则
进程重启后恢复持久化任务存储、执行状态和故障恢复机制
多实例共同处理任务领取协议、租约或确认、重复执行与幂等控制
生产监控接收与拒绝计数、排队耗时、执行耗时、失败与 panic 计数

当前实现中，一个被接收的任务体最多调用一次；开始前观察到取消时可以一次也不调用。这个进程内约定不能推出业务副作用“恰好发生一次”，也不能提供跨进程的 exactly-once 保证。

真正需要守住的基础是：入口能够控制负载，执行数量有明确上限，已接收任务有结果，关闭协议没有发送与关闭 channel 的竞态。把这些边界写进代码和测试，后续加入重试、定时或持久化时才有可靠的起点。