threshold

README

Threshold Executor

Threshold Executor是一个基于阈值的批量执行器。它支持当数据积累到一定阈值或周期到达时触发执行。

特性

• 数量阈值触发执行
• 时间周期触发执行
• 数量阈值和周期触发可配置
• 支持指定批量执行和周期执行回调函数
• 自动批量处理数据
• 阈值达到时主动推送
• 可配置恢复处理函数

用法

executor := NewThresholdExecutor(handler)

• handler为每次执行的回调函数

配置

可配置批量阈值、周期等:

executor.WithBatchSize(1000)  
executor.WithPeriodic(time.Second)

也可分别配置数量阈值和周期的回调函数:

executor.WithBatchSizeHandler(batchHandler)
executor.WithPeriodicHandler(periodicHandler)

收集数据

executor.Collect(data)

达到阈值时会自动执行处理。

执行

executor.AsyncExecute() 

启动执行器。

控制

executor.Stop()

可安全停止执行器。

欢迎提意见或PR帮助改进!

Documentation

Index

• type ThresholdExecute
  • func NewThresholdExecute[ITEM any](itemDo func(i int, item ITEM)) *ThresholdExecute[ITEM]
  • func (exec *ThresholdExecute[ITEM]) AsyncExecute() *ThresholdExecute[ITEM]
  • func (exec *ThresholdExecute[ITEM]) Collect(item ITEM)
  • func (pe *ThresholdExecute[ITEM]) Stop()
  • func (pe *ThresholdExecute[ITEM]) WithBatchSize(bsize int) *ThresholdExecute[ITEM]
  • func (pe *ThresholdExecute[ITEM]) WithBatchSizeHandler(itemSizeDo func(i int, item ITEM)) *ThresholdExecute[ITEM]
  • func (pe *ThresholdExecute[ITEM]) WithPeriodic(per time.Duration) *ThresholdExecute[ITEM]
  • func (pe *ThresholdExecute[ITEM]) WithPeriodicHandler(itemPeriodicDo func(i int, item ITEM)) *ThresholdExecute[ITEM]
  • func (pe *ThresholdExecute[ITEM]) WithRecover(recoverDo func(ierr any)) *ThresholdExecute[ITEM]

Types

type ThresholdExecute

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

ThresholdExecute 阈值执行, 超过阈值就执行. 必须调用AsyncExecute才能执行. 默认 batchsize 128 periodic 100ms

func NewThresholdExecute

func NewThresholdExecute[ITEM any](itemDo func(i int, item ITEM)) *ThresholdExecute[ITEM]

func (*ThresholdExecute[ITEM]) AsyncExecute

func (exec *ThresholdExecute[ITEM]) AsyncExecute() *ThresholdExecute[ITEM]

AsyncExecute 返回自身. 方便与With设置连用

func (*ThresholdExecute[ITEM]) Collect

func (exec *ThresholdExecute[ITEM]) Collect(item ITEM)

Collect 收集数据

func (*ThresholdExecute[ITEM]) Stop

func (pe *ThresholdExecute[ITEM]) Stop()

Stop 停止执行

func (*ThresholdExecute[ITEM]) WithBatchSize

func (pe *ThresholdExecute[ITEM]) WithBatchSize(bsize int) *ThresholdExecute[ITEM]

func (*ThresholdExecute[ITEM]) WithBatchSizeHandler

func (pe *ThresholdExecute[ITEM]) WithBatchSizeHandler(itemSizeDo func(i int, item ITEM)) *ThresholdExecute[ITEM]

func (*ThresholdExecute[ITEM]) WithPeriodic

func (pe *ThresholdExecute[ITEM]) WithPeriodic(per time.Duration) *ThresholdExecute[ITEM]

func (*ThresholdExecute[ITEM]) WithPeriodicHandler

func (pe *ThresholdExecute[ITEM]) WithPeriodicHandler(itemPeriodicDo func(i int, item ITEM)) *ThresholdExecute[ITEM]

func (*ThresholdExecute[ITEM]) WithRecover

func (pe *ThresholdExecute[ITEM]) WithRecover(recoverDo func(ierr any)) *ThresholdExecute[ITEM]