Help us shape the future of Go! Take the 2026 Go Developer Survey

engine

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 10, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Overview

Package engine 实现所有执行器共享的执行引擎.

引擎是"采集与执行解耦"的落点:

  • Collect/Notify 只负责把数据放入有界队列(可配置阻塞或丢弃);
  • 唯一的 dispatcher goroutine 拥有全部可变状态, 热路径上不需要互斥锁;
  • 到点(或达到阈值)时把批次交给 Handler, 并按需限制并发;
  • Close 幂等; Wait 可同步等待在途任务与残留数据全部结束;
  • 周期、阈值、并发数等参数支持运行期热更新, 下一轮生效.

该包属于内部实现, 不应被业务代码直接导入.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Backpressure 

type Backpressure int

Backpressure 决定队列/并发槽位不足时的行为. 

const (
	// BackpressureBlock 阻塞等待(背压), 不丢数据. 默认值.
	BackpressureBlock Backpressure = iota
	// BackpressureDrop 丢弃当前数据并计数, 不阻塞调用方.
	BackpressureDrop
)

func (Backpressure) String 

func (b Backpressure) String() string

type Collector 

type Collector[ITEM any] struct {
	// contains filtered or unexported fields
}

Collector 是采集句柄, 是"采集与执行解耦"的载体.

采集方只需要持有 *Collector 即可持续投递数据, 完全不需要知道执行周期、 阈值、并发度与生命周期; 引擎关闭后所有句柄统一返回 basic.ErrClosed.

句柄本身无状态、可复制(值语义), 可在多个 goroutine 之间共享.

func (*Collector[ITEM]) Engine 

func (c *Collector[ITEM]) Engine() *Engine[ITEM]

Engine 返回句柄所属的引擎, 便于访问指标与生命周期方法.

func (*Collector[ITEM]) Push 

func (c *Collector[ITEM]) Push(item ITEM) error

Push 收集一条数据. 队列满时按引擎的背压策略处理(默认阻塞等待).

func (*Collector[ITEM]) PushAll 

func (c *Collector[ITEM]) PushAll(items ...ITEM) error

PushAll 批量收集. 返回第一条失败的错误, 已成功入队的数据不会被回滚.

func (*Collector[ITEM]) TryPush 

func (c *Collector[ITEM]) TryPush(item ITEM) error

TryPush 非阻塞收集一条数据; 队列满时返回 basic.ErrQueueFull.

type Config 

type Config[ITEM any] struct {
	// Handler 是默认处理器.
	Handler basic.Handler[ITEM]
	// PeriodicHandler 在"周期兜底"路径覆盖 Handler(可选).
	PeriodicHandler basic.Handler[ITEM]
	// SizeHandler 在"数量阈值"路径覆盖 Handler(可选).
	SizeHandler basic.Handler[ITEM]

	// Mode 是执行时机.
	Mode Mode
	// Interval 是执行周期.
	Interval time.Duration
	// BatchSize 是数量阈值, 仅 ModeThreshold 使用; <=0 表示禁用数量触发.
	BatchSize int
	// Concurrency 是并发执行批次的 worker 数量; <=0 时取 runtime.NumCPU().
	Concurrency int
	// QueueSize 是采集队列容量; <=0 时取默认值 65536.
	QueueSize int
	// BufferCap 是批次缓冲区初始容量; <=0 时取 BatchSize 或 1024.
	BufferCap int

	// Backpressure 决定背压策略.
	Backpressure Backpressure
	// MaxPending 限制缓冲区积压上限; <=0 表示不限制. 超出即丢弃并计数.
	MaxPending int
	// DeferOnThreshold 仅 ModeConcurrent 有效: 为 true 时达到数量阈值立即派发,
	// 而不是等下一个节拍(threshold 模式内部会打开它).
	DeferOnThreshold bool

	// Recover 在 Handler panic 时被调用(可选).
	Recover basic.RecoverFunc
	// OnExecute 每次批次执行完成后同步回调(必须轻量).
	OnExecute func(basic.ExecuteEvent)
	// OnDrop 每次丢弃时同步回调(必须轻量).
	OnDrop func(basic.DropEvent)
	// Logf 是内部诊断日志(默认静默).
	Logf func(format string, args ...any)

	// RecoverKeep 是保留的最近失败记录条数; <=0 时取默认值 8.
	RecoverKeep int
	// Name 用于日志与指标标识.
	Name string
}

Config 是执行引擎的完整配置.

创建期固定: Handler、PeriodicHandler、SizeHandler、Mode、QueueSize、BufferCap。 运行期可热更新(见 Options/Configure): Interval、BatchSize、Concurrency、 Backpressure、MaxPending、Recover、OnExecute、OnDrop、Logf。

type Engine 

type Engine[ITEM any] struct {
	// contains filtered or unexported fields
}

Engine 是执行引擎.

除生命周期状态与指标外, 所有业务可变状态都由 dispatcher 独占: buffer 只被 dispatcher 与 Flush/Close 的"取走"路径触碰, 由 owners 保护.

func New 

func New[ITEM any](cfg Config[ITEM], opts ...Option[ITEM]) (*Engine[ITEM], error)

New 创建执行引擎, 并在首次采集时自动启动执行循环.

func (*Engine[ITEM]) Await 

func (e *Engine[ITEM]) Await() basic.Stats

Await 阻塞直到 dispatcher 处理完此前提交的所有数据, 并返回其后的指标快照.

典型用法: 批量投递数据后调用 Await, 确保指标(如 DroppedItems)已经结算. 注意: 它只保证"命令已处理完成", 若节拍触发的批次正在并发执行, 可能仍有 InFlight 中的批次未结束, 此时请配合 Wait 使用.

func (*Engine[ITEM]) Close 

func (e *Engine[ITEM]) Close()

Close 优雅关闭: 幂等, 不阻塞等待执行结束.

关闭语义(重要):

  • 停止接收新数据(之后的 Collect 返回 ErrClosed);
  • 已经进入队列的数据会先被合并进缓冲区, 再按背压策略执行: 阻塞策略下会一直等到执行完(绝不静默丢数据), 丢弃策略下未处理的会计入 DroppedItems;
  • 因此不存在"数据凭空消失": 任意时刻都满足 ReceivedItems == ProcessedItems + DroppedItems + PendingItems。

需要同步等待全部结束请使用 Wait。

func (*Engine[ITEM]) Collect 

func (e *Engine[ITEM]) Collect(item ITEM) error

Collect 收集一条数据(默认句柄); 队列满时按背压策略阻塞, 关闭后返回 ErrClosed.

func (*Engine[ITEM]) Collector 

func (e *Engine[ITEM]) Collector() *Collector[ITEM]

Collector 返回一个独立的采集句柄.

采集与执行解耦的关键: 句柄可在多个 goroutine / 子系统之间传递, 各自持有一份 引用互不干扰; 引擎关闭后句柄立即失效(返回 ErrClosed), 无需额外的同步协议.

func (*Engine[ITEM]) Configure 

func (e *Engine[ITEM]) Configure(opts ...Option[ITEM])

Configure 应用运行期配置(热更新). 传入的 Option 立即生效: 周期在下一个网格点生效, 阈值/并发数等在下一轮生效, 不丢数据、不需要重启.

func (*Engine[ITEM]) Done 

func (e *Engine[ITEM]) Done() <-chan struct{}

Done 返回在 dispatcher 完全退出后关闭的通道.

func (*Engine[ITEM]) Flush 

func (e *Engine[ITEM]) Flush() error

Flush 让当前缓冲区中的数据立即执行, 并等待执行完成.

常驻循环的执行器一般不需要它; 主要用于 ModeTrigger 的多生产者场景 (把一批数据手动刷出去)以及测试.

func (*Engine[ITEM]) FlushAndDrain 

func (e *Engine[ITEM]) FlushAndDrain() error

FlushAndDrain 语义同 Flush, 但在执行器已关闭时改为同步执行残留数据(尽力而为).

func (*Engine[ITEM]) Pause 

func (e *Engine[ITEM]) Pause()

Pause 暂停执行: 仍可收集数据, 但不会触发处理.

该调用会等待状态真正切换完成(dispatcher 处理到该命令)后返回, 因此 "Pause 返回后新采集的数据一定不会被自动执行". 被暂停前已经入队的数据仍会保留在缓冲区中, Resume 后继续处理.

func (*Engine[ITEM]) Pending 

func (e *Engine[ITEM]) Pending() int

Pending 返回缓冲区中待处理的数据条数.

func (*Engine[ITEM]) Resume 

func (e *Engine[ITEM]) Resume()

Resume 从暂停中恢复; 返回后周期性执行重新生效.

func (*Engine[ITEM]) Start 

func (e *Engine[ITEM]) Start()

Start 启动 dispatcher; 重复调用无副作用.

func (*Engine[ITEM]) Stats 

func (e *Engine[ITEM]) Stats() basic.Stats

Stats 返回运行指标快照.

func (*Engine[ITEM]) Status 

func (e *Engine[ITEM]) Status() basic.Status

Status 返回生命周期状态.

func (*Engine[ITEM]) TryCollect 

func (e *Engine[ITEM]) TryCollect(item ITEM) error

TryCollect 非阻塞收集; 队列满时返回 ErrQueueFull 并计数.

func (*Engine[ITEM]) Wait 

func (e *Engine[ITEM]) Wait(ctx context.Context) error

Wait 阻塞等待执行器完全停止, ctx 用于超时控制.

type Mode 

type Mode int

Mode 决定执行时机. 

const (
	// ModeInterval 固定节奏: 每隔 Interval 把缓冲区中的数据作为一个批次执行.
	ModeInterval Mode = iota
	// ModeCompensate 固定速率补偿: 批次执行耗时超过 Interval 时, 下一轮不额外等待,
	// 使长期平均速率稳定在 1/Interval.
	ModeCompensate
	// ModeConcurrent 并发批量: 每隔 Interval 取走缓冲区中的数据, 交给最多 Concurrency
	// 个 worker 并发执行.
	ModeConcurrent
	// ModeThreshold 阈值触发: 数据条数达到 BatchSize 立即执行; Interval 作为兜底,
	// 定期把不足阈值的数据也执行掉.
	ModeThreshold
	// ModeTrigger 被动触发: 不使用定时器, 每次数据到达时把当前缓冲区中的数据
	// 合并成一个批次立即执行.
	ModeTrigger
)

func (Mode) String 

func (m Mode) String() string

type Option 

type Option[ITEM any] func(*Options[ITEM])

Option 是对运行期配置的修改, 可安全地在运行期重复应用.

func WithBackpressure 

func WithBackpressure[ITEM any](bp Backpressure) Option[ITEM]

WithBackpressure 设置背压策略.

func WithBatchSize 

func WithBatchSize[ITEM any](size int) Option[ITEM]

WithBatchSize 设置数量阈值(size > 0).

func WithConcurrency 

func WithConcurrency[ITEM any](n int) Option[ITEM]

WithConcurrency 设置并发 worker 数量.

func WithDeferOnThreshold 

func WithDeferOnThreshold[ITEM any](v bool) Option[ITEM]

WithDeferOnThreshold 仅在 ModeConcurrent 下有意义: 为 true 时达到数量阈值 立即派发批次, 而不是等下一个节拍. 使用它必须同时提供 BatchSize.

func WithInterval 

func WithInterval[ITEM any](d time.Duration) Option[ITEM]

WithInterval 设置执行周期(d > 0), 运行期热更新在下一个网格点生效.

func WithLogf 

func WithLogf[ITEM any](fn func(format string, args ...any)) Option[ITEM]

WithLogf 设置内部诊断日志.

func WithMaxPending 

func WithMaxPending[ITEM any](n int) Option[ITEM]

WithMaxPending 设置缓冲区积压上限; n <= 0 表示不限制.

func WithOnDrop 

func WithOnDrop[ITEM any](fn func(basic.DropEvent)) Option[ITEM]

WithOnDrop 设置丢弃回调.

func WithOnExecute 

func WithOnExecute[ITEM any](fn func(basic.ExecuteEvent)) Option[ITEM]

WithOnExecute 设置批次执行完成回调.

func WithPeriodicHandler 

func WithPeriodicHandler[ITEM any](h basic.Handler[ITEM]) Option[ITEM]

WithPeriodicHandler 设置周期兜底处理器.

func WithRecover 

func WithRecover[ITEM any](r basic.RecoverFunc) Option[ITEM]

WithRecover 设置 panic 恢复回调.

func WithSizeHandler 

func WithSizeHandler[ITEM any](h basic.Handler[ITEM]) Option[ITEM]

WithSizeHandler 设置数量阈值处理器.

type Options 

type Options[ITEM any] struct {
	// Interval / SetInterval 表示是否显式设置了周期; 未设置时保留原值.
	Interval    time.Duration
	SetInterval bool

	BatchSize int
	SetSize   bool

	Backpressure    Backpressure
	SetBackpressure bool

	Concurrency    int
	SetConcurrency bool

	MaxPending    int
	SetMaxPending bool

	Recover   basic.RecoverFunc
	OnExecute func(basic.ExecuteEvent)
	OnDrop    func(basic.DropEvent)
	Logf      func(format string, args ...any)

	// DeferOnThreshold 是创建期语义: 仅当 SetDeferOnThreshold 为 true 时生效.
	DeferOnThreshold    bool
	SetDeferOnThreshold bool

	// PeriodicHandler / SizeHandler 为空表示保留原处理器.
	PeriodicHandler basic.Handler[ITEM]
	SizeHandler     basic.Handler[ITEM]
}

Options 是运行期可热更新的配置子集.

与 Config 分离的原因: WithInterval(0) 这类调用必须能区分 "未设置" 与 "显式设置为零值"(后者用于停用某个触发条件).

type Trigger 

type Trigger uint8

Trigger 说明一次批次执行的来源, 决定 threshold 模式下使用哪个处理器. 

const (
	// TriggerPeriodic 周期性兜底(节拍到点)或显式 Flush.
	TriggerPeriodic Trigger = iota
	// TriggerThreshold 数据量达到数量阈值.
	TriggerThreshold
	// TriggerArrival 数据到达即执行(ModeTrigger).
	TriggerArrival
)

func (Trigger) String 

func (t Trigger) String() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.