batch

Documentation

Overview

Package batch 提供批量任务的有界并发、热更新分类容量、集群共享租约和异常隔离能力。

Index

• func Do(ctx context.Context, work func(context.Context) error) error
• func Map[T, R any](ctx context.Context, items []T, options Options, ...) ([]Result[R], Summary, error)
• func MapObserved[T, R any](ctx context.Context, items []T, options Options, ...) ([]Result[R], Summary, error)
• func Run[T any](ctx context.Context, items []T, options Options, ...) ([]Result[struct{}], Summary, error)
• type LeaseLimiter
• type Options
• type PanicError
  • func (e *PanicError) Error() string
• type Pool
  • func NewChildPool(limit int, parent *Pool) *Pool
  • func NewPool(limit int) *Pool
  • func NewSharedChildPool(limit int, limiter LeaseLimiter, key string, parent *Pool) *Pool
  • func NewSharedPool(limit int, limiter LeaseLimiter, key string) *Pool
  • func (p *Pool) Do(ctx context.Context, work func(context.Context) error) (err error)
  • func (p *Pool) Limit() int
  • func (p *Pool) Snapshot() PoolSnapshot
  • func (p *Pool) UpdateJitter(maximum time.Duration)
  • func (p *Pool) UpdateLimit(limit int)
• type PoolSnapshot
• type Result
• type Summary
  • func ForEachObserved[T, R any](ctx context.Context, items []T, options Options, ...) (Summary, error)

Functions

func Do

func Do(ctx context.Context, work func(context.Context) error) error

Do 隔离单个任务 panic，适用于长驻 Worker 和后台任务监督器。

func Map

func Map[T, R any](ctx context.Context, items []T, options Options, work func(context.Context, T) (R, error)) ([]Result[R], Summary, error)

Map 以稳定输入顺序返回结果；单项失败和 panic 不会中断其他已提交任务。

func MapObserved

func MapObserved[T, R any](ctx context.Context, items []T, options Options, work func(context.Context, T) (R, error), observe func(index int, result Result[R])) ([]Result[R], Summary, error)

MapObserved 在任务释放共享容量后通知结果观察者，适合连接下游有界流水线。

func Run

func Run[T any](ctx context.Context, items []T, options Options, work func(context.Context, T) error) ([]Result[struct{}], Summary, error)

Run 执行只关心成功或失败的批量任务。

Types

type LeaseLimiter

type LeaseLimiter interface {
	Acquire(ctx context.Context, key string, limit int) (release func(), acquired bool, err error)
}

type Options

type Options struct {
	Workers   int
	QueueSize int
	Pool      *Pool
}

type PanicError

type PanicError struct {
	Value any
	Stack []byte
}

PanicError 表示任务发生 panic；堆栈只用于服务端诊断，不应直接返回给客户端。

func (*PanicError) Error

func (e *PanicError) Error() string

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

Pool 在多个批量操作之间共享并发容量，避免并发请求叠加突破上游保护阈值。

func NewChildPool

func NewChildPool(limit int, parent *Pool) *Pool

NewChildPool 创建分类并发池；任务先取得分类容量，再进入父级全局池。

func NewPool

func NewPool(limit int) *Pool

func NewSharedChildPool

func NewSharedChildPool(limit int, limiter LeaseLimiter, key string, parent *Pool) *Pool

NewSharedChildPool 创建同时受分类集群租约和父级总容量约束的并发池。

func NewSharedPool

func NewSharedPool(limit int, limiter LeaseLimiter, key string) *Pool

func (*Pool) Do

func (p *Pool) Do(ctx context.Context, work func(context.Context) error) (err error)

Do 等待共享执行容量并隔离任务 panic，调用方仍通过 context 控制排队和执行生命周期。

func (*Pool) Limit

func (p *Pool) Limit() int

Limit 返回当前并发上限。

func (*Pool) Snapshot

func (p *Pool) Snapshot() PoolSnapshot

func (*Pool) UpdateJitter

func (p *Pool) UpdateJitter(maximum time.Duration)

UpdateJitter 热更新任务进入并发池前的随机延迟上限；零表示关闭。

func (*Pool) UpdateLimit

func (p *Pool) UpdateLimit(limit int)

UpdateLimit 热更新并发上限；降低上限不会中断正在执行的任务。

type PoolSnapshot

type PoolSnapshot struct {
	Limit  int
	Active int
	Queued int
	Peak   int
}

type Result

type Result[T any] struct {
	Value     T
	Err       error
	Completed bool
}

type Summary

type Summary struct {
	Total     int
	Submitted int
	Completed int
	Succeeded int
	Failed    int
	Panicked  int
	Canceled  bool
	Duration  time.Duration
}

func ForEachObserved

func ForEachObserved[T, R any](ctx context.Context, items []T, options Options, work func(context.Context, T) (R, error), observe func(index int, result Result[R])) (Summary, error)

ForEachObserved 执行带观察者的批量任务，但不在内存中保留全部结果。 适用于单项结果已经流式发送给下游、最终只需要汇总信息的批量接口。