periodic

README

Periodic Executor

Periodic Executor是一个可配置的周期性任务执行器,支持多种执行模式。

特性

• 定时执行模式
  • 固定间隔循环执行
  • 执行时间补偿模式
  • 定期批量并发模式
• 数据收集与执行解耦
• 错误处理及恢复机制
• 执行控制
  • 支持启动、停止
  • 执行状态跟踪
• 高性能
  • 利用channel、锁、原子操作实现并发安全

用法

创建执行器

executor := NewExecuteInterval(handlerFunc)

传入执行函数。

配置

可配置执行周期、错误恢复等:

executor.WithPeriodic(dur)
executor.WithRecover(recoverFunc)

数据收集

executor.Collect(data)

分离收集和执行。

执行

executor.AsyncExecute() 

支持三种模式:

• IntervalLoop - 固定间隔循环执行
• Compensate - 执行时间补偿模式
• Concurrent - 定期批量并发模式

控制

executor.Stop() // 停止执行

自定义执行器

实现 Executor 接口即可实现自定义执行器集成使用。

TODO

• 增加详细文档
• 支持动态配置更改

欢迎报告issue或提交PR!

Documentation

Index

• type ConcurrentExecute
  • func NewConcurrentExecute[ITEM any](execDo func(i int, item ITEM)) *ConcurrentExecute[ITEM]
  • func (pe *ConcurrentExecute[ITEM]) AsyncExecute() *ConcurrentExecute[ITEM]
  • func (pe *ConcurrentExecute[ITEM]) Collect(item ITEM)
  • func (pe *ConcurrentExecute[ITEM]) Stop()
  • func (pe *ConcurrentExecute[ITEM]) WithPeriodic(per time.Duration) *ConcurrentExecute[ITEM]
  • func (pe *ConcurrentExecute[ITEM]) WithRecover(recoverDo func(ierr any)) *ConcurrentExecute[ITEM]
• type ExecuteCompensate
  • func NewExecuteCompensate[ITEM any](execDo func(i int, item ITEM)) *ExecuteCompensate[ITEM]
  • func (pe *ExecuteCompensate[ITEM]) AsyncExecute() *ExecuteCompensate[ITEM]
  • func (pe *ExecuteCompensate[ITEM]) Collect(item ITEM)
  • func (pe *ExecuteCompensate[ITEM]) Stop()
  • func (pe *ExecuteCompensate[ITEM]) WithPeriodic(per time.Duration) *ExecuteCompensate[ITEM]
  • func (pe *ExecuteCompensate[ITEM]) WithRecover(recoverDo func(ierr any)) *ExecuteCompensate[ITEM]
• type ExecuteInterval
  • func NewExecuteInterval[ITEM any](execDo func(i int, item ITEM)) *ExecuteInterval[ITEM]
  • func (pe *ExecuteInterval[ITEM]) AsyncExecute() *ExecuteInterval[ITEM]
  • func (pe *ExecuteInterval[ITEM]) Collect(item ITEM)
  • func (pe *ExecuteInterval[ITEM]) Stop()
  • func (pe *ExecuteInterval[ITEM]) WithPeriodic(per time.Duration) *ExecuteInterval[ITEM]
  • func (pe *ExecuteInterval[ITEM]) WithRecover(recoverDo func(ierr any)) *ExecuteInterval[ITEM]

Types

type ConcurrentExecute

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

ConcurrentExecute 定期并发

func NewConcurrentExecute

func NewConcurrentExecute[ITEM any](execDo func(i int, item ITEM)) *ConcurrentExecute[ITEM]

func (*ConcurrentExecute[ITEM]) AsyncExecute

func (pe *ConcurrentExecute[ITEM]) AsyncExecute() *ConcurrentExecute[ITEM]

func (*ConcurrentExecute[ITEM]) Collect

func (pe *ConcurrentExecute[ITEM]) Collect(item ITEM)

Collect 收集数据

func (*ConcurrentExecute[ITEM]) Stop

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

Stop 停止执行

func (*ConcurrentExecute[ITEM]) WithPeriodic

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

func (*ConcurrentExecute[ITEM]) WithRecover

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

type ExecuteCompensate

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

ExecuteCompensate 时间补偿执行

func NewExecuteCompensate

func NewExecuteCompensate[ITEM any](execDo func(i int, item ITEM)) *ExecuteCompensate[ITEM]

func (*ExecuteCompensate[ITEM]) AsyncExecute

func (pe *ExecuteCompensate[ITEM]) AsyncExecute() *ExecuteCompensate[ITEM]

func (*ExecuteCompensate[ITEM]) Collect

func (pe *ExecuteCompensate[ITEM]) Collect(item ITEM)

Collect 收集数据

func (*ExecuteCompensate[ITEM]) Stop

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

Stop 停止执行

func (*ExecuteCompensate[ITEM]) WithPeriodic

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

func (*ExecuteCompensate[ITEM]) WithRecover

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

type ExecuteInterval

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

ExecuteInterval 时间间隔执行

func NewExecuteInterval

func NewExecuteInterval[ITEM any](execDo func(i int, item ITEM)) *ExecuteInterval[ITEM]

func (*ExecuteInterval[ITEM]) AsyncExecute

func (pe *ExecuteInterval[ITEM]) AsyncExecute() *ExecuteInterval[ITEM]

func (*ExecuteInterval[ITEM]) Collect

func (pe *ExecuteInterval[ITEM]) Collect(item ITEM)

Collect 收集数据

func (*ExecuteInterval[ITEM]) Stop

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

Stop 停止执行

func (*ExecuteInterval[ITEM]) WithPeriodic

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

func (*ExecuteInterval[ITEM]) WithRecover

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