basic

type Context ¶ added in v0.3.0

type Context = context.Context

Context 是 context.Context 的别名, 让上层无需重复导入 context.

type DropEvent ¶ added in v0.3.0

type DropEvent struct {
	// At 是丢弃时间.
	At time.Time
	// Reason 是丢弃原因, 例如 "queue-full"、"max-pending"、"not-running".
	Reason string
	// PendingItems 是丢弃时刻缓冲区中的积压条数.
	PendingItems int
}

DropEvent 描述一条因背压策略被丢弃的数据, 用于 OnDrop 观测回调.

type Duration ¶ added in v0.3.0

type Duration = time.Duration

Duration 是 time.Duration 的别名, 便于上层在不引入 time 包的情况下书写回调签名.

type ErrorRecord ¶ added in v0.3.0

type ErrorRecord struct {
	// At 是失败发生时间.
	At time.Time
	// Items 是本次失败批次的数据条数.
	Items int
	// Err 是失败描述(Handler 返回的 error 或 panic 的格式化文本).
	Err string
	// Panicked 表示失败来自 panic 而不是 error 返回值.
	Panicked bool
}

ErrorRecord 描述一次失败的执行.

type ExecuteEvent ¶ added in v0.3.0

type ExecuteEvent struct {
	// Items 是本次批次的数据条数.
	Items int
	// Err 是失败原因, nil 表示成功.
	Err error
	// Panicked 表示失败来自 panic 而非 error 返回值.
	Panicked bool
	// Duration 是 Handler 的执行耗时.
	Duration Duration
	// Time 是执行开始时间.
	Time time.Time
}

ExecuteEvent 描述一次批次执行的结果, 用于 OnExecute 观测回调.

该回调在执行路径上同步调用, 必须保持轻量(仅做计数、打点或投递到独立队列), 否则会拖慢执行吞吐.

func (e ExecuteEvent) OK() bool

type Executor ¶ added in v0.3.0

type Executor[ITEM any] interface {
	// Collect 收集一条数据. 当内部缓冲区满时可能阻塞(背压), 关闭后返回错误.
	Collect(item ITEM) error
	// TryCollect 非阻塞地收集一条数据, 缓冲区满时返回 ErrQueueFull.
	TryCollect(item ITEM) error
	// Flush 阻塞地等待缓冲区中被当前批次遗漏的数据执行完成.
	Flush() error
	// Start 显式启动执行循环(通常无需调用, Collect 会自动启动).
	Start()
	// Pause 暂停执行: 仍可收集数据, 但不会触发处理.
	Pause()
	// Resume 从暂停中恢复.
	Resume()
	// Close 优雅关闭: 停止接收新数据, 等待在途任务与残留数据执行完成.
	// 该方法是幂等的, 且不阻塞调用方等待执行结束(需要同步等待请使用 Wait).
	Close()
	// Wait 阻塞等待执行器完全停止, ctx 用于超时控制.
	// 返回 nil 表示已完全停止, 返回 ctx.Err() 表示等待超时.
	Wait(ctx Context) error
	// Status 返回当前生命周期状态.
	Status() Status
	// Stats 返回运行指标快照.
	Stats() Stats
}

Executor 是所有执行器的公共契约.

生命周期:

Created --Start/Collect--> Running --Pause--> Paused --Resume--> Running
Running/Paused --Close--> Closing --(在途任务结束)--> Stopped

实现必须保证:

• 所有已导出的方法都可并发调用;
• Close 幂等;
• 关闭后 Collect 不再 panic, 而是返回错误.

type Handler ¶ added in v0.3.0

type Handler[ITEM any] func(items []ITEM) error

Handler 是执行器真正执行的业务逻辑. items 是本次批次的数据.

约定:

• items 由执行器内部复用, 仅在本次调用内有效; 如果需要在调用返回后继续持有, 必须自行复制(例如 slices.Clone);
• 返回 error 表示本批次处理失败, 执行器会记录指标并继续运行, 不会中断执行循环;
• panic 会被执行器捕获并按失败处理, 执行循环不会因此退出.

之所以定义为命名类型(而非匿名函数类型), 是为了让执行器内部结构体能够以 "已实例化的类型"保存回调, 从而不必在方法上重复声明类型参数. 匿名函数可直接赋值或转换:

var h basic.Handler[Item] = func(items []Item) error { return nil }

type RecoverFunc ¶

type RecoverFunc func(ierr any)

RecoverFunc 在 Handler panic 时被调用, 用于日志、告警、上报等副作用. ierr 即 recover() 的返回值. 执行器自身仍会记录失败指标, 因此 RecoverFunc 只应做轻量、不可再 panic 的处理.

type Semaphore ¶ added in v0.3.0

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

Semaphore 是一个可动态调整上限的计数信号量, 用于限制并发度.

三个概念要区分清楚:

limit —— 当前并发上限, 可通过 Resize 在运行期调整;
token —— 目前可用的 permit 数(缓冲通道内的元素个数);
inUse —— limit - token, 即在途持有者数量。

不变式: 0 <= token <= limit <= cap(slots)。cap(slots) 是创建时申请的 "允许扩容到的最大并发度", limit 则是运行期实际生效的上限。

所有方法都可并发调用.

func NewSemaphore(n int) *Semaphore

func (s *Semaphore) Acquire(ctx context.Context) (bool, error)

func (s *Semaphore) AcquireWait() bool

func (s *Semaphore) Available() int

func (s *Semaphore) Cap() int

func (s *Semaphore) InUse() int

func (s *Semaphore) Limit() int

func (s *Semaphore) Release()

func (s *Semaphore) Resize(n int) int

func (s *Semaphore) TryAcquire() bool

type Stats ¶ added in v0.3.0

type Stats struct {
	ReceivedItems    uint64
	ProcessedItems   uint64
	SucceededItems   uint64
	FailedItems      uint64
	DroppedItems     uint64
	Executions       uint64
	FailedExecutions uint64

	// PendingItems 是快照时刻仍在缓冲区中等待的数据条数.
	PendingItems int
	// InFlight 是快照时刻正在并发执行的批次数量.
	InFlight int64

	// LastError / LastErrorAt 记录最近一次失败的描述与时间.
	LastError   string
	LastErrorAt time.Time
	// RecentErrors 记录最近若干次失败, 便于排障.
	RecentErrors []ErrorRecord
}

Stats 是执行器运行指标的只读快照.

计数均为累计值, 语义如下:

• ReceivedItems: 成功进入执行器缓冲区的数据条数(不含被丢弃的);
• ProcessedItems: 交给 Handler 处理过的数据条数;
• SucceededItems: 处理过程中未产生错误/panic 的数据条数;
• FailedItems: 因 Handler 返回 error 或 panic 而未成功的数据条数;
• DroppedItems: 因背压策略(丢弃模式/超出上限)而未被接收的数据条数;
• Executions: Handler 被调用的批次数;
• FailedExecutions: 其中失败的批次数.

func (s Stats) ErrorRate() float64

type Status ¶ added in v0.3.0

type Status int32

Status 描述执行器的生命周期状态.

const (
	// StatusCreated 已创建, 尚未启动(启动是惰性的, 首次收集或显式 Start 时进入 Running).
	StatusCreated Status = iota
	// StatusRunning 正在运行, 可以接收数据并执行.
	StatusRunning
	// StatusPaused 已暂停: 仍然接收数据, 但不会触发执行.
	StatusPaused
	// StatusClosing 正在关闭: 不再接收新数据, 等待在途任务结束.
	StatusClosing
	// StatusStopped 已停止, 所有资源已释放.
	StatusStopped
)

func (s Status) Runnable() bool

func (s Status) String() string

func (s Status) Terminal() bool

type Time ¶ added in v0.3.0

type Time = time.Time

Time 是 time.Time 的别名, 避免基础包暴露多余导入.