Documentation
¶
Overview ¶
Package batch 提供攒批执行器：把零散任务攒成批次按条数/字节数/时间触发 批量执行，核心是 PeriodicalExecutor + TaskContainer 扩展点。
设计参考 go-zero core/executors（PeriodicalExecutor 的按需后台协程 + 空闲自退出模型），差异点：
- Ticker/时钟可注入，批量时序测试完全确定（无需真实睡眠）；
- Execute 统一经 async.RunSafe 隔离 panic，一批失败不影响后续；
- 变体精简为 Bulk（满 N 条）/ Chunk（满字节）/ Less（窗口去重）三个高频件。
常用形态：
// 攒满 64 条或每 100ms 批量下发
exe := batch.NewBulkExecutor(100*time.Millisecond, 64, func(tasks []any) {
flush(tasks)
})
exe.Add(task)
defer exe.Wait() // 进程退出前排空积压
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BatchOption ¶
type BatchOption func(*batchOptions)
BatchOption 攒批执行器的可选参数。
func WithBatchClock ¶
func WithBatchClock(now func() time.Time) BatchOption
WithBatchClock 注入时间源（LessExecutor 等按墙钟判定的执行器使用）。 nil 函数被忽略。
func WithBatchTicker ¶
func WithBatchTicker(f TickerFactory) BatchOption
WithBatchTicker 注入周期信号源工厂（测试用手动 Ticker 即可确定性地驱动 flush 时序，无需真实睡眠）。nil 工厂被忽略。
type BulkExecutor ¶
type BulkExecutor struct {
*PeriodicalExecutor
}
BulkExecutor 满条数触发版攒批执行器：攒满 maxTasks 条立即 flush， 否则等 interval 周期。interval <= 0 表示只按条数触发。
func NewBulkExecutor ¶
func NewBulkExecutor(interval time.Duration, maxTasks int, do func(tasks []any), opts ...BatchOption) *BulkExecutor
NewBulkExecutor 创建满条数触发的攒批执行器。maxTasks <= 0 或 do 为 nil 属于编程错误，直接 panic。
type ChunkExecutor ¶
type ChunkExecutor struct {
*PeriodicalExecutor
}
ChunkExecutor 满字节触发版攒批执行器：按任务声明的大小累计， 攒满 maxBytes 立即 flush，否则等 interval 周期。适合按负载体量 （日志字节、消息字节数）控制批次大小的场景。
func NewChunkExecutor ¶
func NewChunkExecutor(interval time.Duration, maxBytes int64, do func(tasks []any), opts ...BatchOption) *ChunkExecutor
NewChunkExecutor 创建满字节触发的攒批执行器。maxBytes <= 0 或 do 为 nil 属于编程错误，直接 panic。
type LessExecutor ¶
type LessExecutor struct {
// contains filtered or unexported fields
}
LessExecutor 窗口去重执行器：同一 interval 窗口内至多执行一次， 其余调用丢弃。适合告警/日志限频——同一事件风暴只放行第一条。
func NewLessExecutor ¶
func NewLessExecutor(interval time.Duration, do func(), opts ...BatchOption) *LessExecutor
NewLessExecutor 创建窗口去重执行器。interval <= 0 或 do 为 nil 属于 编程错误，直接 panic。
func (*LessExecutor) DoOrDiscard ¶
func (e *LessExecutor) DoOrDiscard() bool
DoOrDiscard 窗口内首次调用执行 do 并返回 true，其余丢弃返回 false。 并发调用下同一窗口保证恰有一路执行（CAS 裁决）。
type PeriodicalExecutor ¶
type PeriodicalExecutor struct {
// contains filtered or unexported fields
}
PeriodicalExecutor 攒批执行器：把零散任务攒进 TaskContainer，按间隔 周期或容器要求立即批量执行。
并发模型：
- 首次 Add 按需拉起唯一后台循环；连续 idleExitRounds 个 interval 无积压 时自退出，下次 Add 重新拉起（退出与 Add 在同一临界区内判定，任务不会 滞留无人消费）；
- 取走积压（RemoveAll）在锁内完成，Execute 在锁外串行执行且 panic 被隔离；
- Add/Flush/Wait 均可在多 goroutine 并发调用。
func NewPeriodicalExecutor ¶
func NewPeriodicalExecutor(interval time.Duration, container TaskContainer, opts ...BatchOption) *PeriodicalExecutor
NewPeriodicalExecutor 创建攒批执行器。interval 为批量触发周期（<=0 表示 不设周期，仅容器主动要求或 Flush/Wait 驱动）；container 为攒批容器。
func (*PeriodicalExecutor) Add ¶
func (e *PeriodicalExecutor) Add(task any)
Add 收纳一个任务；容器要求立即 flush（AddTask 返回 true）时唤醒后台。
func (*PeriodicalExecutor) Flush ¶
func (e *PeriodicalExecutor) Flush()
Flush 同步排空：立即取走当前积压并在调用方 goroutine 执行。与后台循环 并发调用安全（取走原子，双方各执行一半也保证不重不漏）。
func (*PeriodicalExecutor) Wait ¶
func (e *PeriodicalExecutor) Wait()
Wait 请求后台排空积压后退出并阻塞至退出完成（幂等；退出后继续 Add 会 重新拉起后台）。适合进程退出前兜底，防止尾部批次丢失。
type TaskContainer ¶
type TaskContainer interface {
// AddTask 收纳一个任务，返回 true 表示需要立即触发一次 flush
// （例如攒满条数或字节数）。
AddTask(task any) bool
// Execute 执行一批任务。执行器保证同一执行器同一时刻至多一路调用
// （后台循环串行），panic 由执行器隔离，实现方无需自行 recover。
Execute(tasks []any)
// RemoveAll 取走并清空当前积压的全部任务；在执行器内部锁内调用，
// 实现方无需自行加锁。
RemoveAll() []any
}
TaskContainer 攒批容器扩展点：执行器只负责"何时触发 flush"， "怎么攒、怎么执行"由容器实现决定。Bulk/Chunk 变体均构建其上。
Source Files