syncx

func (*AtomicValue[T]) CompareAndSwap ¶

func (av *AtomicValue[T]) CompareAndSwap(old, new T) bool

CompareAndSwap 原子地比较并交换值

func (*AtomicValue[T]) Load ¶

func (av *AtomicValue[T]) Load() T

Load 原子地加载值

func (*AtomicValue[T]) Store ¶

func (av *AtomicValue[T]) Store(val T)

Store 原子地存储值

func (*AtomicValue[T]) Swap ¶

func (av *AtomicValue[T]) Swap(new T) T

Swap 原子地交换值并返回旧值

func (*BatchExecutor) Context ¶

func (b *BatchExecutor) Context() context.Context

Context 返回执行器的 context

func (*BatchExecutor) ErrorCount ¶

func (b *BatchExecutor) ErrorCount() int

ErrorCount 获取错误总数（并发安全）

func (*BatchExecutor) Errors ¶

func (b *BatchExecutor) Errors() map[int64]error

Errors 获取所有错误映射（并发安全） 返回 map[taskID]error，其中 taskID 是任务提交的顺序编号（从1开始）

func (*BatchExecutor) Go ¶

func (b *BatchExecutor) Go(fn func() error)

Go 提交一个任务（并发安全）

func (*BatchExecutor) HasErrors ¶

func (b *BatchExecutor) HasErrors() bool

HasErrors 检查是否有错误（并发安全）

func (*BatchExecutor) OnError ¶

func (b *BatchExecutor) OnError(fn GoExecutorErrorHandler) *BatchExecutor

OnError 设置错误处理器（每个错误都会调用）

func (*BatchExecutor) OnPanic ¶

func (b *BatchExecutor) OnPanic(fn GoExecutorPanicHandler) *BatchExecutor

OnPanic 设置 panic 处理器

func (*BatchExecutor) SetLimit ¶

func (b *BatchExecutor) SetLimit(n int) *BatchExecutor

SetLimit 设置最大并发数

func (*BatchExecutor) SetMode ¶

func (b *BatchExecutor) SetMode(mode BatchExecutorMode) *BatchExecutor

SetMode 设置执行模式

func (*BatchExecutor) Wait ¶

func (b *BatchExecutor) Wait() error

Wait 等待所有任务完成，返回第一个错误（并发安全）

func (*BatchProcessor[T]) DroppedCount ¶

func (p *BatchProcessor[T]) DroppedCount() int64

DroppedCount 返回累计丢弃的条目数（队列满时丢弃） 用于监控背压情况，判断是否需要扩容 queueSize 或优化 flush 性能

func (*BatchProcessor[T]) Name ¶

func (p *BatchProcessor[T]) Name() string

Name 返回处理器名称（通过 WithName 设置）

func (*BatchProcessor[T]) Stop ¶

func (p *BatchProcessor[T]) Stop()

Stop 停止处理器，flush 剩余数据后退出

func (*BatchProcessor[T]) Submit ¶

func (p *BatchProcessor[T]) Submit(item T) bool

Submit 非阻塞提交一条数据

行为说明：

• 队列未满：item 写入 channel，返回 true
• 队列已满：不会阻塞等待，也不会重试，直接走 default 分支丢弃该 item，返回 false
• 若设置了 WithClone，Submit 时会先克隆 item（数据隔离）

也就是说队列满之后，新提交的数据进不来，会被静默丢弃（最终一致性语义） 调用方可根据返回值决定后续处理：

• 记录 metric（如 DroppedCount）便于监控背压
• 降级处理（如改走同步更新、写本地缓冲等）

这种"非阻塞 + 丢弃"策略适用于可丢失的非核心路径（如消息状态更新）， 避免生产者因队列满而阻塞，拖垮上游主流程

func (*BatchProcessor[T]) SubmitBlocking ¶

func (p *BatchProcessor[T]) SubmitBlocking(ctx context.Context, item T) bool

SubmitBlocking 阻塞式提交一条数据（不丢数据的兜底方案）

行为说明：

• 队列未满：item 立即写入 channel，返回 true
• 队列已满：阻塞等待，直到有空位可写入（返回 true）或 ctx 被取消/超时（返回 false）
• 若设置了 WithClone，Submit 时会先克隆 item（数据隔离）

与 Submit 的区别：

• Submit：队列满立即丢弃，适合可丢失的非核心路径（如消息状态更新）
• SubmitBlocking：队列满时背压传导给调用方，适合不能丢数据的核心路径（如订单、扣款）

注意事项：

• 阻塞期间会占用调用 goroutine，建议配合 ctx 超时使用，避免长时间卡住
• 若下游 flush 持续慢导致队列长期满，阻塞会向上游传导背压，上游需有相应限流/超时机制
• ctx.Done() 触发时该条数据被丢弃（由调用方决定是否重试/落盘补偿）

兜底策略选择：

• 不想丢 + 可接受短暂阻塞 → SubmitBlocking(ctx, item)
• 不想阻塞 + 可接受丢失 → Submit(item)
• 想要"先阻塞再降级" → SubmitBlocking 配合较短 ctx 超时，超时后走降级逻辑

func (*Bool) CAS ¶

func (b *Bool) CAS(old, new bool) bool

CAS 原子地比较并交换布尔值

func (*Bool) Load ¶

func (b *Bool) Load() bool

Load 原子地加载布尔值

func (*Bool) Store ¶

func (b *Bool) Store(val bool)

Store 原子地存储布尔值

func (*Bool) Toggle ¶

func (b *Bool) Toggle() bool

Toggle 原子地切换布尔值

func (*ChildRunner) Go ¶

func (c *ChildRunner) Go(fn GoExecutorFunc) *ChildRunner

Go 启动一个子 goroutine

func (*ChildRunner) GoWithError ¶

func (c *ChildRunner) GoWithError(fn GoExecutorWaitFunc) *ChildRunner

GoWithError 启动一个带错误返回的子 goroutine

func (*ChildRunner) Wait ¶

func (c *ChildRunner) Wait()

Wait 等待所有子 goroutine 完成

func (*Delayer[T]) Close ¶

func (d *Delayer[T]) Close()

Close 关闭结果通道

func (*Delayer[T]) Execute ¶

func (d *Delayer[T]) Execute() error

func (*Delayer[T]) GetLegacyResults ¶

func (d *Delayer[T]) GetLegacyResults() []*ExecutionResult

GetLegacyResults 获取所有执行结果（向后兼容）

func (*Delayer[T]) GetResultChannel ¶

func (d *Delayer[T]) GetResultChannel() <-chan T

GetResultChannel 获取结果通道（只读）

func (*Delayer[T]) GetResults ¶

func (d *Delayer[T]) GetResults() []T

GetResults 获取所有泛型任务结果

func (*Delayer[T]) GetStats ¶

func (d *Delayer[T]) GetStats() *ExecutionStats

GetStats 获取执行统计

func (*Delayer[T]) IsRunning ¶

func (d *Delayer[T]) IsRunning() bool

IsRunning 检查是否正在运行

func (*Delayer[T]) Stop ¶

func (d *Delayer[T]) Stop()

Stop 停止执行

func (*Delayer[T]) Wait ¶

func (d *Delayer[T]) Wait() error

Wait 等待上下文取消

func (*Delayer[T]) WaitForCompletion ¶

func (d *Delayer[T]) WaitForCompletion()

WaitForCompletion 等待任务完成

func (*Delayer[T]) WithConcurrent ¶

func (d *Delayer[T]) WithConcurrent(concurrent bool) *Delayer[T]

WithConcurrent 设置是否并发执行

func (*Delayer[T]) WithContext ¶

func (d *Delayer[T]) WithContext(ctx context.Context) *Delayer[T]

WithContext 设置上下文

func (*Delayer[T]) WithCustomDelay ¶

func (d *Delayer[T]) WithCustomDelay(delayFunc DelayFunc) *Delayer[T]

WithCustomDelay 设置自定义延迟函数

func (*Delayer[T]) WithDelay ¶

func (d *Delayer[T]) WithDelay(delay time.Duration) *Delayer[T]

WithDelay 设置基础延迟时间

func (*Delayer[T]) WithDisableCallbacks ¶

func (d *Delayer[T]) WithDisableCallbacks(disable bool) *Delayer[T]

WithDisableCallbacks 设置是否禁用回调以提升性能

func (*Delayer[T]) WithFunction ¶

func (d *Delayer[T]) WithFunction(f func() error) *Delayer[T]

WithFunction 设置要执行的函数

func (*Delayer[T]) WithMaxConcurrency ¶

func (d *Delayer[T]) WithMaxConcurrency(maxConcurrency int) *Delayer[T]

WithMaxConcurrency 设置最大并发数

func (*Delayer[T]) WithMaxDelay ¶

func (d *Delayer[T]) WithMaxDelay(maxDelay time.Duration) *Delayer[T]

WithMaxDelay 设置最大延迟时间

func (*Delayer[T]) WithMultiplier ¶

func (d *Delayer[T]) WithMultiplier(multiplier float64) *Delayer[T]

WithMultiplier 设置指数策略的倍数

func (*Delayer[T]) WithOnAfterComplete ¶

func (d *Delayer[T]) WithOnAfterComplete(callback func(ctx *ExecutionContext)) *Delayer[T]

WithOnAfterComplete 设置完成执行后回调

func (*Delayer[T]) WithOnBeforeStart ¶

func (d *Delayer[T]) WithOnBeforeStart(callback func(ctx *ExecutionContext)) *Delayer[T]

WithOnBeforeStart 设置开始执行前回调

func (*Delayer[T]) WithOnComplete ¶

func (d *Delayer[T]) WithOnComplete(callback func(result *ExecutionResult)) *Delayer[T]

WithOnComplete 设置完成执行回调（向后兼容）

func (*Delayer[T]) WithOnError ¶

func (d *Delayer[T]) WithOnError(callback func(index int, err error) bool) *Delayer[T]

WithOnError 设置错误处理回调（向后兼容）

func (*Delayer[T]) WithOnErrorContext ¶

func (d *Delayer[T]) WithOnErrorContext(callback ErrorHandlerFunc) *Delayer[T]

WithOnErrorContext 设置错误处理回调

func (*Delayer[T]) WithOnProgress ¶

func (d *Delayer[T]) WithOnProgress(callback TaskProgressFunc) *Delayer[T]

WithOnProgress 设置进度回调

func (*Delayer[T]) WithOnStart ¶

func (d *Delayer[T]) WithOnStart(callback func(index int)) *Delayer[T]

WithOnStart 设置开始执行回调（向后兼容）

func (*Delayer[T]) WithOnSuccess ¶

func (d *Delayer[T]) WithOnSuccess(callback CallbackFunc[T]) *Delayer[T]

WithOnSuccess 设置成功回调

func (*Delayer[T]) WithRandomBase ¶

func (d *Delayer[T]) WithRandomBase(base float64) *Delayer[T]

WithRandomBase 设置随机基数（用于随机延迟策略）

func (*Delayer[T]) WithSimpleFunction ¶

func (d *Delayer[T]) WithSimpleFunction(f func()) *Delayer[T]

WithSimpleFunction 设置无返回值的函数

func (*Delayer[T]) WithSimpleTaskFunc ¶

func (d *Delayer[T]) WithSimpleTaskFunc(taskFunc SimpleTaskFunc) *Delayer[T]

WithSimpleTaskFunc 设置简单任务函数（无返回值）

func (*Delayer[T]) WithStopOnError ¶

func (d *Delayer[T]) WithStopOnError(stopOnError bool) *Delayer[T]

WithStopOnError 设置遇到错误是否停止

func (*Delayer[T]) WithStrategy ¶

func (d *Delayer[T]) WithStrategy(strategy DelayStrategy) *Delayer[T]

WithStrategy 设置延迟策略

func (*Delayer[T]) WithTaskFunc ¶

func (d *Delayer[T]) WithTaskFunc(taskFunc TaskFunc[T]) *Delayer[T]

WithTaskFunc 设置泛型任务函数

func (*Delayer[T]) WithTimes ¶

func (d *Delayer[T]) WithTimes(count int) *Delayer[T]

WithTimes 设置执行次数

func (*DynamicTicker) GetInterval ¶

func (dt *DynamicTicker) GetInterval() time.Duration

GetInterval 获取当前间隔

func (*DynamicTicker) Start ¶

func (dt *DynamicTicker) Start()

Start 启动定时器

功能：

• 创建底层 time.Ticker 并启动后台 goroutine
• 如果已经启动，重复调用不会有任何效果

注意：

• 线程安全，可以在多个 goroutine 中调用
• 启动后会立即开始按照 interval 发送 tick 事件

func (*DynamicTicker) Stop ¶

func (dt *DynamicTicker) Stop()

Stop 停止定时器

func (*DynamicTicker) UpdateInterval ¶

func (dt *DynamicTicker) UpdateInterval(interval time.Duration)

UpdateInterval 动态更新定时器间隔 功能：

• 如果新间隔与当前间隔相同，不做任何操作
• 如果定时器正在运行，会立即应用新的间隔
• 使用 ticker.Reset() 实现无缝切换，不会丢失正在进行的 tick

注意：

• 线程安全，可以在任何时候调用
• 可以在定时器启动前或运行中调用

func (*EventLoop) IfChannel ¶

func (el *EventLoop) IfChannel(condition bool, ch interface{}, handler interface{}) *EventLoop

IfChannel 条件注册通道事件处理器 只有当条件为 true 时才注册通道监听

参数:

• condition: 条件判断，为 true 时才注册
• ch: 必须是一个通道类型
• handler: 必须是一个函数，接受通道元素类型的参数

示例:

// 只有当启用某功能时才监听该通道
loop.IfChannel(config.EnableFeature, featureChan, func(msg Message) {
    handleFeature(msg)
})

func (*EventLoop) IfTicker ¶

func (el *EventLoop) IfTicker(condition bool, interval time.Duration, handler func()) *EventLoop

IfTicker 条件注册定时器事件处理器 只有当条件为 true 时才注册定时器

参数:

• condition: 条件判断，为 true 时才注册
• interval: 定时间隔（如 5*time.Second, 1*time.Minute）
• handler: 定时触发时执行的函数，无参数无返回值

示例:

// 只有当启用清理时才注册定时器
loop.IfTicker(config.EnableCleanup, 30*time.Minute, func() {
    cleanupExpiredData()
})

func (*EventLoop) OnChannel ¶

func (el *EventLoop) OnChannel(ch interface{}, handler interface{}) *EventLoop

OnChannel 注册通道事件处理器 当通道接收到数据时，会调用对应的处理函数

参数:

• ch: 必须是一个通道类型（如 chan int, chan string, chan *Client 等）
• handler: 必须是一个函数，接受通道元素类型的参数

类型检查:

• 会在运行时通过反射验证 ch 是否为通道类型
• 会验证 handler 的参数类型是否匹配通道元素类型

示例:

registerChan := make(chan *Client, 100)
loop.OnChannel(registerChan, func(client *Client) {
    handleRegister(client)
})

messageChan := make(chan Message, 100)
loop.OnChannel(messageChan, func(msg Message) {
    handleMessage(msg)
})

func (*EventLoop) OnPanic ¶

func (el *EventLoop) OnPanic(fn func(interface{})) *EventLoop

OnPanic 设置 panic 处理器 当事件处理函数中发生 panic 时，会调用此函数

特性:

• 捕获所有处理函数中的 panic，防止整个事件循环崩溃
• panic 发生后，事件循环会继续运行，处理其他事件
• 如果设置了多次，只有最后一次设置的函数会生效

用途:

• 记录 panic 日志
• 上报错误到监控系统
• 发送告警通知

示例:

loop.OnPanic(func(r interface{}) {
    log.Error("事件处理panic", "panic", r, "stack", string(debug.Stack()))
    sentry.CaptureException(fmt.Errorf("panic: %v", r))
})

func (*EventLoop) OnShutdown ¶

func (el *EventLoop) OnShutdown(fn func()) *EventLoop

OnShutdown 设置关闭时的回调 当事件循环退出时（context 取消或发生 panic），会调用此函数进行清理

用途:

• 关闭数据库连接
• 清理临时文件
• 保存状态数据
• 记录关闭日志

注意:

• 此函数会在 defer 中执行，保证一定会被调用
• 如果设置了多次，只有最后一次设置的函数会生效

示例:

loop.OnShutdown(func() {
    log.Info("事件循环已停止")
    db.Close()
    saveState()
})

func (*EventLoop) OnTicker ¶

func (el *EventLoop) OnTicker(interval time.Duration, handler func()) *EventLoop

OnTicker 注册定时器事件处理器 按指定的时间间隔周期性地执行处理函数

参数:

• interval: 定时间隔（如 5*time.Second, 1*time.Minute）
• handler: 定时触发时执行的函数，无参数无返回值

特性:

• 定时器会在 Run() 方法中自动创建
• 定时器会在事件循环结束时自动停止
• 支持注册多个定时器，它们互不干扰

示例:

// 每5秒检查一次心跳
loop.OnTicker(5*time.Second, func() {
    checkHeartbeat()
})

// 每1分钟清理一次过期数据
loop.OnTicker(1*time.Minute, func() {
    cleanupExpiredData()
})

func (*EventLoop) Run ¶

func (el *EventLoop) Run()

Run 运行事件循环（阻塞） 开始监听所有注册的事件源，直到 context 被取消

执行流程:

1. 构建 select cases（包括通道、定时器）
2. 启动所有定时器
3. 进入 select 循环，等待事件
4. 接收到事件时，调用对应的处理函数
5. context 取消时退出循环
6. 停止所有定时器，调用 OnShutdown 回调

错误处理:

• 自动捕获所有处理函数中的 panic
• panic 不会导致事件循环崩溃
• 通过 OnPanic 回调通知上层

示例:

loop := NewEventLoop(ctx).
    OnChannel(ch1, handler1).
    OnChannel(ch2, handler2).
    OnTicker(5*time.Second, tickHandler)

// 阻塞运行，直到 context 被取消
loop.Run()

func (*EventLoop) RunAsync ¶

func (el *EventLoop) RunAsync()

RunAsync 异步运行事件循环（非阻塞） 在新的 goroutine 中运行事件循环，立即返回

与 Run 的区别:

• Run: 阻塞当前 goroutine，直到 context 被取消
• RunAsync: 在新 goroutine 中运行，立即返回

使用场景:

• 需要在主 goroutine 中继续执行其他任务
• 需要同时运行多个事件循环

示例:

loop := NewEventLoop(ctx).
    OnChannel(ch, handler).
    OnTicker(5*time.Second, tickHandler)

// 异步运行，不阻塞
loop.RunAsync()

// 可以继续执行其他代码
doOtherWork()

func (*ExecutionResult) ToExecutionContext ¶

func (r *ExecutionResult) ToExecutionContext() *ExecutionContext

ToExecutionContext 将 ExecutionResult 转换为 ExecutionContext

func (*FuncChain[T]) AddFuncItem ¶

func (fc *FuncChain[T]) AddFuncItem(item *FuncItem[T]) *FuncChain[T]

AddFuncItem 将 FuncItem 添加到 FuncChain 中

func (*FuncChain[T]) Clear ¶

func (fc *FuncChain[T]) Clear()

Clear 清空 FuncChain 中的所有 FuncItem

func (*FuncChain[T]) Execute ¶

func (fc *FuncChain[T]) Execute()

Execute 按顺序执行所有添加的函数，使用协程

func (*FuncChain[T]) GetFuncItems ¶

func (fc *FuncChain[T]) GetFuncItems() []FuncItem[T]

GetFuncItems 返回 FuncChain 中的所有 FuncItem（副本）

func (*FuncItem[T]) GetError ¶

func (fi *FuncItem[T]) GetError() error

GetError 获取 FuncItem Error

func (*FuncItem[T]) GetResult ¶

func (fi *FuncItem[T]) GetResult() T

GetResult 获取 FuncItem Result

func (*FuncItem[T]) WithPriority ¶

func (fi *FuncItem[T]) WithPriority(priority int) *FuncItem[T]

WithPriority 设置 FuncItem 的优先级

func (*GoExecutor) Exec ¶

func (g *GoExecutor) Exec(fn GoExecutorFunc)

Exec 执行无参数的函数

示例:

Go().OnPanic(handler).Exec(func() {
    doSomething()
})

func (*GoExecutor) ExecSubs ¶

func (g *GoExecutor) ExecSubs()

ExecSubs 执行所有子 goroutine 任务并等待完成

示例:

Go().OnPanic(handler).
    Sub(func() { task1() }).
    Sub(func() { task2() }).
    SubWithError(func() error { return task3() }).
    ExecSubs()

func (*GoExecutor) ExecWithChildren ¶

func (g *GoExecutor) ExecWithChildren(fn func(*ChildRunner))

ExecWithChildren 在父 goroutine 中执行，并管理多个子 goroutine

示例:

Go().OnPanic(handler).ExecWithChildren(func(children *ChildRunner) {
    children.Go(func() { startHeartbeat() })
    children.Go(func() { subscribe1() })
    children.Go(func() { subscribe2() })
})

func (*GoExecutor) ExecWithContext ¶

func (g *GoExecutor) ExecWithContext(fn GoExecutorContextFunc)

ExecWithContext 执行带 Context 的函数

示例:

Go().WithTimeout(5*time.Second).OnError(handler).ExecWithContext(func(ctx context.Context) error {
    return repo.Save(ctx, data)
})

func (*GoExecutor) ExecWithResult ¶

func (g *GoExecutor) ExecWithResult(fn GoExecutorResultFunc) <-chan struct {
	Value interface{}
	Err   error
}

ExecWithResult 执行带返回值的函数(异步获取结果)

示例:

resultChan := Go().ExecWithResult(func() (int, error) {
    return computeValue(), nil
})
select {
case result := <-resultChan:
    if result.Err != nil { ... }
    value := result.Value
}

func (*GoExecutor) OnCancel ¶

func (g *GoExecutor) OnCancel(fn GoExecutorCancelHandler) *GoExecutor

OnCancel 设置取消回调

示例:

Go().OnCancel(func() { log.Info("cancelled") }).ExecWithContext(fn)

func (*GoExecutor) OnError ¶

func (g *GoExecutor) OnError(fn GoExecutorErrorHandler) *GoExecutor

OnError 设置错误回调

示例:

Go().OnError(func(err error) { log.Error(err) }).ExecWithContext(fn)

func (*GoExecutor) OnPanic ¶

func (g *GoExecutor) OnPanic(fn GoExecutorPanicHandler) *GoExecutor

OnPanic 设置 panic 回调

示例:

Go().OnPanic(func(r interface{}) { log.Error("panic", r) }).Exec(fn)

func (*GoExecutor) Sub ¶

func (g *GoExecutor) Sub(fn GoExecutorFunc) *GoExecutor

Sub 添加子 goroutine 任务（链式调用）

示例:

Go().OnPanic(handler).
    Sub(func() { startHeartbeat() }).
    Sub(func() { subscribe1() }).
    Sub(func() { subscribe2() }).
    ExecSubs()

func (*GoExecutor) SubWithError ¶

func (g *GoExecutor) SubWithError(fn GoExecutorWaitFunc) *GoExecutor

SubWithError 添加带错误返回的子 goroutine 任务（链式调用）

func (*GoExecutor) Wait ¶

func (g *GoExecutor) Wait(fn GoExecutorWaitFunc) error

Wait 同步等待执行完成

示例:

Go().Wait(func() error {
    return doSomething()
})

func (*GoExecutor) WithDelay ¶

func (g *GoExecutor) WithDelay(delay time.Duration) *GoExecutor

WithDelay 设置延迟执行时间

示例:

Go().WithDelay(2*time.Second).Exec(fn)

func (*GoExecutor) WithTimeout ¶

func (g *GoExecutor) WithTimeout(timeout time.Duration) *GoExecutor

WithTimeout 设置超时时间

示例:

Go().WithTimeout(5*time.Second).ExecWithContext(fn)

func (*GoExecutor) WithWaitGroup ¶

func (g *GoExecutor) WithWaitGroup(wg *sync.WaitGroup) *GoExecutor

WithWaitGroup 绑定外部 WaitGroup，goroutine 启动前 Add(1)，完成后 Done() 用于需要等待异步任务完成的场景（如优雅关闭时等待所有写入完成）

注意：调用方无需手动 Add(1)，Exec 系列方法会自动管理计数

示例:

var wg sync.WaitGroup
Go().WithWaitGroup(&wg).WithTimeout(5*time.Second).ExecWithContext(func(ctx context.Context) error {
    return repo.Save(ctx, data)
})
wg.Wait() // 等待所有绑定的 goroutine 完成

func (*HashedWheelTimer) Cancel ¶

func (t *HashedWheelTimer) Cancel(tt *TimerTask)

Cancel 取消指定任务（惰性取消，O(1)）

func (*HashedWheelTimer) CancelByKey ¶

func (t *HashedWheelTimer) CancelByKey(key string)

CancelByKey 按 key 取消任务（惰性取消，O(1)）

func (*HashedWheelTimer) Refresh ¶

func (t *HashedWheelTimer) Refresh(key string, delay time.Duration, task func()) *TimerTask

Refresh 刷新任务（O(1) 心跳刷新：取消旧 + 调度新）

func (*HashedWheelTimer) Schedule ¶

func (t *HashedWheelTimer) Schedule(delay time.Duration, task func()) *TimerTask

Schedule 调度一个无 key 的定时任务（轮询分配 shard）

func (*HashedWheelTimer) ScheduleWithKey ¶

func (t *HashedWheelTimer) ScheduleWithKey(key string, delay time.Duration, task func()) *TimerTask

ScheduleWithKey 调度一个带 key 的定时任务（同 key 覆盖会惰性取消旧任务）

func (*HashedWheelTimer) Stats ¶

func (t *HashedWheelTimer) Stats() TimerStats

Stats 返回时间轮统计信息 completed/cancelled 为各 shard 计数器之和（下沉到 shard 避免全局原子竞争）

func (*HashedWheelTimer) Stop ¶

func (t *HashedWheelTimer) Stop()

Stop 停止时间轮（停止所有 worker，不再执行待触发任务）

func (*Int32) Add ¶

func (i32 *Int32) Add(i int32) int32

Add 原子地增加值

func (*Int32) CAS ¶

func (i32 *Int32) CAS(old, new int32) bool

CAS 原子地比较并交换值

func (*Int32) Load ¶

func (i32 *Int32) Load() int32

Load 原子地加载值

func (*Int32) Store ¶

func (i32 *Int32) Store(i int32)

Store 原子地存储值

func (*Int32) String ¶

func (i32 *Int32) String() string

String 返回值的字符串表示

func (*Int32) Sub ¶

func (i32 *Int32) Sub(i int32) int32

Sub 原子地减少值

func (*Int32) Swap ¶

func (i32 *Int32) Swap(i int32) int32

Swap 原子地交换值

func (*Int64) Add ¶

func (i64 *Int64) Add(i int64) int64

Add 原子地增加值

func (*Int64) CAS ¶

func (i64 *Int64) CAS(old, new int64) bool

CAS 原子地比较并交换值

func (*Int64) Load ¶

func (i64 *Int64) Load() int64

Load 原子地加载值

func (*Int64) Store ¶

func (i64 *Int64) Store(i int64)

Store 原子地存储值

func (*Int64) String ¶

func (i64 *Int64) String() string

String 返回值的字符串表示

func (*Int64) Sub ¶

func (i64 *Int64) Sub(i int64) int64

Sub 原子地减少值

func (*Int64) Swap ¶

func (i64 *Int64) Swap(i int64) int64

Swap 原子地交换值

func (*LimitedPool) Get ¶

func (p *LimitedPool) Get(size int) *[]byte

Get 从池中获取指定大小的字节切片

func (*LimitedPool) Put ¶

func (p *LimitedPool) Put(b *[]byte)

Put 将字节切片放回池中以供重用

func (*Lock) Lock ¶

func (l *Lock) Lock()

Lock 获取锁

func (*Lock) TryLock ¶

func (l *Lock) TryLock() bool

TryLock 尝试获取锁

func (*Lock) Unlock ¶

func (l *Lock) Unlock()

Unlock 释放锁

func (*Map[K, V]) All ¶

func (m *Map[K, V]) All(predicate func(K, V) bool) bool

All 判断是否所有元素都满足条件

func (*Map[K, V]) Any ¶

func (m *Map[K, V]) Any(predicate func(K, V) bool) bool

Any 判断是否存在满足条件的元素

func (*Map[K, V]) Clear ¶

func (m *Map[K, V]) Clear()

Clear 清空 Map 中的所有键值对

func (*Map[K, V]) Clone ¶

func (m *Map[K, V]) Clone() *Map[K, V]

Clone 克隆当前 Map 实例，返回一个新的 Map 实例

func (*Map[K, V]) CompareAndDelete ¶

func (m *Map[K, V]) CompareAndDelete(key K, value V) bool

CompareAndDelete 比较指定键的值，如果相等则删除该键的值。 使用 sync.Map 的原生原子方法，避免 Load+Delete 之间的 TOCTOU 竞争

func (*Map[K, V]) CompareAndSwap ¶

func (m *Map[K, V]) CompareAndSwap(key K, old, new V) bool

CompareAndSwap 比较指定键的现有值，如果相等则将其替换为新值。 使用 sync.Map 的原生原子方法，避免 Load+Store 之间的 TOCTOU 竞争

func (*Map[K, V]) Count ¶

func (m *Map[K, V]) Count(predicate func(K, V) bool) int

Count 返回满足条件的元素数量

func (*Map[K, V]) Delete ¶

func (m *Map[K, V]) Delete(key K)

Delete 删除指定键的值。

func (*Map[K, V]) DeleteIf ¶

func (m *Map[K, V]) DeleteIf(predicate func(K, V) bool) int

DeleteIf 删除满足条件的所有元素

func (*Map[K, V]) Equals ¶

func (m *Map[K, V]) Equals(key K, value V, cmpFunc func(existing V) bool) bool

Equals 函数用于比较两个键值对是否相等， 需要用户提供自定义的比较函数。

func (*Map[K, V]) Filter ¶

func (m *Map[K, V]) Filter(predicate func(K, V) bool) []V

Filter 过滤 Map，返回满足条件的元素组成的新切片

func (*Map[K, V]) FilterKeys ¶

func (m *Map[K, V]) FilterKeys(predicate func(K, V) bool) []K

FilterKeys 过滤 Map，返回满足条件的键组成的切片

func (*Map[K, V]) FilterMap ¶

func (m *Map[K, V]) FilterMap(predicate func(K, V) bool) *Map[K, V]

FilterMap 过滤 Map，返回满足条件的元素组成的新 Map

func (*Map[K, V]) ForEach ¶

func (m *Map[K, V]) ForEach(fn func(K, V))

ForEach 遍历 Map 并对每个元素执行操作

func (*Map[K, V]) GetOrCompute ¶

func (m *Map[K, V]) GetOrCompute(key K, compute func() V) V

GetOrCompute 获取键对应的值，如果不存在则计算并存储

func (*Map[K, V]) GetOrStore ¶

func (m *Map[K, V]) GetOrStore(key K, defaultValue V) V

GetOrStore 获取键对应的值，如果不存在则存储并返回默认值

func (*Map[K, V]) IsEmpty ¶

func (m *Map[K, V]) IsEmpty() bool

IsEmpty 判断 Map 是否为空

func (*Map[K, V]) Keys ¶

func (m *Map[K, V]) Keys() []K

Keys 返回所有键的切片

func (*Map[K, V]) Load ¶

func (m *Map[K, V]) Load(key K) (V, bool)

Load 获取指定键的值。

func (*Map[K, V]) LoadAndDelete ¶

func (m *Map[K, V]) LoadAndDelete(key K) (V, bool)

LoadAndDelete 方法从 Map 中加载并删除指定键的值

func (*Map[K, V]) LoadOrStore ¶

func (m *Map[K, V]) LoadOrStore(key K, new V) (V, bool)

LoadOrStore 获取指定键的值，如果不存在则存储新值。

func (*Map[K, V]) Range ¶

func (m *Map[K, V]) Range(run func(key K, value V) bool)

Range 遍历 Map 中的所有键值对。

func (*Map[K, V]) Size ¶

func (m *Map[K, V]) Size() int

Size 返回当前 Map 中元素的数量

func (*Map[K, V]) Store ¶

func (m *Map[K, V]) Store(key K, value V)

Store 设置指定键的值。

func (*Map[K, V]) Swap ¶

func (m *Map[K, V]) Swap(key K, value V) (pre V, ok bool)

Swap 替换指定键的值，并返回之前的值。 使用 sync.Map 的原生原子方法，避免 Load+Delete+Store 之间的竞争

func (*Map[K, V]) ToMap ¶

func (m *Map[K, V]) ToMap() map[K]V

ToMap 将 syncx.Map 转换为普通 map

func (*Map[K, V]) Update ¶

func (m *Map[K, V]) Update(key K, updater func(V) V) bool

Update 更新指定键的值，如果键存在则使用 updater 函数更新，返回是否更新成功

func (*Map[K, V]) Values ¶

func (m *Map[K, V]) Values() []V

Values 返回所有值的切片

func (*OrderedSet[T]) Add ¶

func (s *OrderedSet[T]) Add(item T)

Add 添加元素，若元素已存在，则忽略 Params:

• item T：待添加元素

func (*OrderedSet[T]) Clear ¶

func (s *OrderedSet[T]) Clear()

Clear 清空集合，移除所有元素，重置内部状态

func (*OrderedSet[T]) Contains ¶

func (s *OrderedSet[T]) Contains(item T) bool

Contains 判断元素是否存在 Params:

• item T：待判断元素

Returns:

• bool：存在返回 true，否则 false

func (*OrderedSet[T]) Elements ¶

func (s *OrderedSet[T]) Elements() []T

Elements 返回集合中所有元素的切片，保持插入顺序 Returns:

• []T：元素切片的副本，修改不会影响集合内部状态

func (*OrderedSet[T]) Len ¶

func (s *OrderedSet[T]) Len() int

Len 返回集合中元素个数 Returns:

• int：元素数量

func (*OrderedSet[T]) Remove ¶

func (s *OrderedSet[T]) Remove(item T)

Remove 删除元素，若元素不存在，则无操作 Params:

• item T：待删除元素

func (*OrderedSet[T]) String ¶

func (s *OrderedSet[T]) String() string

String 返回集合的字符串表示，便于打印和调试 Returns:

• string：格式如 OrderedSet{elem1, elem2, ...}

func (*ParallelExecutor[K, V, R]) Execute ¶

func (p *ParallelExecutor[K, V, R]) Execute(fn ParallelExecuteFunc[K, V, R])

Execute 执行并发任务

func (*ParallelExecutor[K, V, R]) OnComplete ¶

func (p *ParallelExecutor[K, V, R]) OnComplete(fn ParallelCompleteCallback[K, R]) *ParallelExecutor[K, V, R]

OnComplete 设置完成回调(所有任务完成后调用)

func (*ParallelExecutor[K, V, R]) OnEachComplete ¶

func (p *ParallelExecutor[K, V, R]) OnEachComplete(fn ParallelEachCompleteCallback[K]) *ParallelExecutor[K, V, R]

OnEachComplete 设置每个任务完成时的回调

func (*ParallelExecutor[K, V, R]) OnError ¶

func (p *ParallelExecutor[K, V, R]) OnError(fn ParallelErrorCallback[K, V]) *ParallelExecutor[K, V, R]

OnError 设置错误回调

func (*ParallelExecutor[K, V, R]) OnPanic ¶

func (p *ParallelExecutor[K, V, R]) OnPanic(fn ParallelPanicCallback[K, V]) *ParallelExecutor[K, V, R]

OnPanic 设置 panic 回调

func (*ParallelExecutor[K, V, R]) OnSuccess ¶

func (p *ParallelExecutor[K, V, R]) OnSuccess(fn ParallelSuccessCallback[K, V, R]) *ParallelExecutor[K, V, R]

OnSuccess 设置成功回调

func (*ParallelSliceExecutor[T, R]) Execute ¶

func (p *ParallelSliceExecutor[T, R]) Execute(fn ParallelSliceExecuteFunc[T, R])

Execute 执行并发任务

工作原理:

• 如果设置了 maxConcurrency，使用信号量控制同时运行的 goroutine 数量
• 所有任务都会被执行，只是控制了并发度
• 使用 WaitGroup 确保所有任务完成后才返回

示例:

• 100个任务，maxConcurrency=10: 同时最多10个 goroutine 在运行，当一个完成后，立即启动下一个 最终所有100个任务都会执行完成

func (*ParallelSliceExecutor[T, R]) OnComplete ¶

func (p *ParallelSliceExecutor[T, R]) OnComplete(fn ParallelSliceCompleteCallback[R]) *ParallelSliceExecutor[T, R]

OnComplete 设置完成回调(所有任务完成后调用)

func (*ParallelSliceExecutor[T, R]) OnEachComplete ¶

func (p *ParallelSliceExecutor[T, R]) OnEachComplete(fn ParallelSliceEachCompleteCallback) *ParallelSliceExecutor[T, R]

OnEachComplete 设置每个任务完成时的回调

func (*ParallelSliceExecutor[T, R]) OnError ¶

func (p *ParallelSliceExecutor[T, R]) OnError(fn ParallelSliceErrorCallback[T]) *ParallelSliceExecutor[T, R]

OnError 设置错误回调

func (*ParallelSliceExecutor[T, R]) OnPanic ¶

func (p *ParallelSliceExecutor[T, R]) OnPanic(fn ParallelSlicePanicCallback[T]) *ParallelSliceExecutor[T, R]

OnPanic 设置 panic 回调

func (*ParallelSliceExecutor[T, R]) OnSuccess ¶

func (p *ParallelSliceExecutor[T, R]) OnSuccess(fn ParallelSliceSuccessCallback[T, R]) *ParallelSliceExecutor[T, R]

OnSuccess 设置成功回调

func (*ParallelSliceExecutor[T, R]) WithConcurrency ¶

func (p *ParallelSliceExecutor[T, R]) WithConcurrency(maxConcurrency int) *ParallelSliceExecutor[T, R]

WithConcurrency 设置最大并发数

参数:

• maxConcurrency: 最大并发 goroutine 数量
• 0: 不限制并发数，所有任务立即启动（默认行为）
• >0: 限制同时运行的 goroutine 数量，使用信号量控制

注意:

• 这只是控制并发数，所有任务最终都会被执行完
• 例如: 100个任务，maxConcurrency=10，表示同时最多10个任务在执行， 当一个任务完成后，会自动启动下一个任务，直到所有100个任务都完成

// 模拟100个任务，但只允许10个并发
tasks := make([]int, 100)
for i := range tasks {
	tasks[i] = i + 1
}

var completed int32

NewParallelSliceExecutor[int, string](tasks).
	WithConcurrency(10). // 限制最多10个并发
	OnSuccess(func(idx int, val int, result string) {
		count := atomic.AddInt32(&completed, 1)
		if count%10 == 0 {
			fmt.Printf("已完成 %d 个任务\n", count)
		}
	}).
	OnComplete(func(results []string, errors []error) {
		fmt.Printf("全部完成！总计 %d 个任务\n", len(results))
	}).
	Execute(func(idx int, val int) (string, error) {
		// 模拟耗时操作
		time.Sleep(time.Millisecond)
		return fmt.Sprintf("任务 %d 完成", val), nil
	})

Output:
已完成 10 个任务
已完成 20 个任务
已完成 30 个任务
已完成 40 个任务
已完成 50 个任务
已完成 60 个任务
已完成 70 个任务
已完成 80 个任务
已完成 90 个任务
已完成 100 个任务
全部完成！总计 100 个任务

func (*PeriodicTask) GetExecuteFunc ¶

func (t *PeriodicTask) GetExecuteFunc() func(ctx context.Context) error

GetExecuteFunc 获取执行函数

func (*PeriodicTask) GetImmediateStart ¶

func (t *PeriodicTask) GetImmediateStart() bool

GetImmediateStart 获取是否立即执行标志

func (*PeriodicTask) GetInterval ¶

func (t *PeriodicTask) GetInterval() time.Duration

GetInterval 获取执行间隔

func (*PeriodicTask) GetName ¶

func (t *PeriodicTask) GetName() string

GetName 获取任务名称

func (*PeriodicTask) GetOnError ¶

func (t *PeriodicTask) GetOnError() func(name string, err error)

GetOnError 获取错误处理回调

func (*PeriodicTask) GetOnOverlapSkipped ¶

func (t *PeriodicTask) GetOnOverlapSkipped() func(name string)

GetOnOverlapSkipped 获取重叠跳过回调

func (*PeriodicTask) GetOnStart ¶

func (t *PeriodicTask) GetOnStart() func(name string)

GetOnStart 获取启动回调

func (*PeriodicTask) GetOnStop ¶

func (t *PeriodicTask) GetOnStop() func(name string)

GetOnStop 获取停止回调

func (*PeriodicTask) GetPreventOverlap ¶

func (t *PeriodicTask) GetPreventOverlap() bool

GetPreventOverlap 获取是否防止重叠执行标志

func (*PeriodicTask) IsExecuting ¶

func (t *PeriodicTask) IsExecuting() bool

IsExecuting 获取当前是否正在执行状态

func (*PeriodicTask) SetImmediateStart ¶

func (t *PeriodicTask) SetImmediateStart(immediateStart bool) *PeriodicTask

SetImmediateStart 设置是否立即执行首次任务

func (*PeriodicTask) SetOnError ¶

func (t *PeriodicTask) SetOnError(onError func(name string, err error)) *PeriodicTask

SetOnError 设置错误处理回调

func (*PeriodicTask) SetOnOverlapSkipped ¶

func (t *PeriodicTask) SetOnOverlapSkipped(onOverlapSkipped func(name string)) *PeriodicTask

SetOnOverlapSkipped 设置重叠跳过回调

func (*PeriodicTask) SetOnStart ¶

func (t *PeriodicTask) SetOnStart(onStart func(name string)) *PeriodicTask

SetOnStart 设置启动回调

func (*PeriodicTask) SetOnStop ¶

func (t *PeriodicTask) SetOnStop(onStop func(name string)) *PeriodicTask

SetOnStop 设置停止回调

func (*PeriodicTask) SetPreventOverlap ¶

func (t *PeriodicTask) SetPreventOverlap(preventOverlap bool) *PeriodicTask

SetPreventOverlap 设置是否防止任务重叠执行

func (*PeriodicTaskManager) AddSimpleTask ¶

func (m *PeriodicTaskManager) AddSimpleTask(name string, interval time.Duration, executeFunc func(ctx context.Context) error) *PeriodicTaskManager

AddSimpleTask 添加简单的周期性任务

func (*PeriodicTaskManager) AddTask ¶

func (m *PeriodicTaskManager) AddTask(task *PeriodicTask) *PeriodicTaskManager

AddTask 添加周期性任务

func (*PeriodicTaskManager) AddTaskWithImmediateStart ¶

func (m *PeriodicTaskManager) AddTaskWithImmediateStart(name string, interval time.Duration, executeFunc func(ctx context.Context) error) *PeriodicTaskManager

AddTaskWithImmediateStart 添加立即执行的周期性任务

func (*PeriodicTaskManager) AddTaskWithOverlapPrevention ¶

func (m *PeriodicTaskManager) AddTaskWithOverlapPrevention(name string, interval time.Duration, executeFunc func(ctx context.Context) error) *PeriodicTaskManager

AddTaskWithOverlapPrevention 添加防重叠执行的周期性任务

func (*PeriodicTaskManager) AddTaskWithOverlapPreventionAndCallback ¶

func (m *PeriodicTaskManager) AddTaskWithOverlapPreventionAndCallback(
	name string,
	interval time.Duration,
	executeFunc func(ctx context.Context) error,
	onOverlapSkipped func(name string),
) *PeriodicTaskManager

AddTaskWithOverlapPreventionAndCallback 添加防重叠执行的周期性任务（带重叠跳过回调）

func (*PeriodicTaskManager) AddTaskWithOverlapPreventionImmediateAndCallback ¶

func (m *PeriodicTaskManager) AddTaskWithOverlapPreventionImmediateAndCallback(
	name string,
	interval time.Duration,
	executeFunc func(ctx context.Context) error,
	onOverlapSkipped func(name string),
) *PeriodicTaskManager

AddTaskWithOverlapPreventionImmediateAndCallback 添加防重叠执行的周期性任务（带立即执行和重叠跳过回调）

func (*PeriodicTaskManager) ClearAllTasks ¶

func (m *PeriodicTaskManager) ClearAllTasks()

ClearAllTasks 清除所有任务

func (*PeriodicTaskManager) GetTaskCount ¶

func (m *PeriodicTaskManager) GetTaskCount() int

GetTaskCount 获取任务数量

func (*PeriodicTaskManager) GetTaskDetails ¶

func (m *PeriodicTaskManager) GetTaskDetails(name ...string) []TaskDetailInfo

GetTaskDetails 获取任务详细信息 如果name为空，返回所有任务；如果指定name，返回匹配的任务

func (*PeriodicTaskManager) GetTaskNames ¶

func (m *PeriodicTaskManager) GetTaskNames() []string

GetTaskNames 获取所有任务名称

func (*PeriodicTaskManager) IsRunning ¶

func (m *PeriodicTaskManager) IsRunning() bool

IsRunning 检查任务管理器是否正在运行

func (*PeriodicTaskManager) RemoveTask ¶

func (m *PeriodicTaskManager) RemoveTask(name string) bool

RemoveTask 移除指定名称的任务

func (*PeriodicTaskManager) RemoveTaskWithTimeout ¶

func (m *PeriodicTaskManager) RemoveTaskWithTimeout(name string, timeout time.Duration) bool

RemoveTaskWithTimeout 移除指定名称的任务（带超时等待）

func (*PeriodicTaskManager) SetDefaultCallbacks ¶

func (m *PeriodicTaskManager) SetDefaultCallbacks(
	onStart func(name string),
	onStop func(name string),
) *PeriodicTaskManager

SetDefaultCallbacks 设置默认回调函数

func (*PeriodicTaskManager) SetDefaultErrorHandler ¶

func (m *PeriodicTaskManager) SetDefaultErrorHandler(handler func(name string, err error)) *PeriodicTaskManager

SetDefaultErrorHandler 设置默认错误处理器

func (*PeriodicTaskManager) Start ¶

func (m *PeriodicTaskManager) Start() error

Start 启动所有周期性任务

func (*PeriodicTaskManager) StartWithContext ¶

func (m *PeriodicTaskManager) StartWithContext(ctx context.Context) error

StartWithContext 使用指定上下文启动所有周期性任务

func (*PeriodicTaskManager) Stop ¶

func (m *PeriodicTaskManager) Stop() error

Stop 停止所有周期性任务

func (*PeriodicTaskManager) StopWithTimeout ¶

func (m *PeriodicTaskManager) StopWithTimeout(timeout time.Duration) error

StopWithTimeout 在指定超时时间内停止所有周期性任务

func (*PeriodicTaskManager) Wait ¶

func (m *PeriodicTaskManager) Wait()

Wait 等待所有任务完成

func (*PeriodicTaskManager) WaitForExecution ¶

func (m *PeriodicTaskManager) WaitForExecution(timeout time.Duration) error

WaitForExecution 等待所有任务至少执行一次 timeout: 最大等待时间，0表示无限等待

func (*Pool[T]) Get ¶

func (p *Pool[T]) Get() T

Get 从池中获取一个对象

func (*Pool[T]) Put ¶

func (p *Pool[T]) Put(x T)

Put 将对象放回池中

func (*RWLock) Lock ¶

func (l *RWLock) Lock()

Lock 获取写锁

func (*RWLock) RLock ¶

func (l *RWLock) RLock()

RLock 获取读锁

func (*RWLock) RUnlock ¶

func (l *RWLock) RUnlock()

RUnlock 释放读锁

func (*RWLock) TryLock ¶

func (l *RWLock) TryLock() bool

TryLock 尝试获取写锁

func (*RWLock) TryRLock ¶

func (l *RWLock) TryRLock() bool

TryRLock 尝试获取读锁

func (*RWLock) Unlock ¶

func (l *RWLock) Unlock()

Unlock 释放写锁

func (*Set[K]) Add ¶

func (s *Set[K]) Add(key K)

Add 向集合中添加一个元素。

func (*Set[K]) AddAll ¶

func (s *Set[K]) AddAll(keys ...K)

AddAll 向集合中添加多个元素。

func (*Set[K]) Clear ¶

func (s *Set[K]) Clear()

Clear 清空集合中的所有元素。

func (*Set[K]) Delete ¶

func (s *Set[K]) Delete(key K)

Delete 从集合中删除指定的元素。

func (*Set[K]) DeleteAll ¶

func (s *Set[K]) DeleteAll(keys ...K)

DeleteAll 从集合中删除多个元素。

func (*Set[K]) Elements ¶

func (s *Set[K]) Elements() []K

Elements 返回集合中的所有元素。

func (*Set[K]) Has ¶

func (s *Set[K]) Has(key K) bool

Has 检查集合中是否存在指定的元素。

func (*Set[K]) HasAll ¶

func (s *Set[K]) HasAll(keys ...K) (existing []K, all bool)

HasAll 检查集合中是否包含所有指定的元素。

func (*Set[K]) IsEmpty ¶

func (s *Set[K]) IsEmpty() bool

IsEmpty 检查集合是否为空。

func (*Set[K]) Size ¶

func (s *Set[K]) Size() int

Size 返回集合中元素的数量。

func (*ShardedMap[K, V]) Clear ¶

func (m *ShardedMap[K, V]) Clear()

Clear 清空所有元素，保留预分配容量提示

func (*ShardedMap[K, V]) ClearHalf ¶

func (m *ShardedMap[K, V]) ClearHalf()

ClearHalf 随机清空每个 shard 内一半的元素

用于大容量缓存的超限驱逐：相比整表 Clear 保留一半数据，避免缓存命中率雪崩 （重建窗口内的查询全部回源），驱逐成本相同（无 LRU 链表维护开销） 每 shard 独立驱逐一半（map 遍历序随机），无轮换状态、多次调用持续有效

func (*ShardedMap[K, V]) CompareAndDelete ¶

func (m *ShardedMap[K, V]) CompareAndDelete(key K, value V) bool

CompareAndDelete 原子比较并删除 仅当 key 存在且当前值等于 value 时，才删除 注意：V 为非可比较类型（如 slice/map）时会 panic，与 sync.Map 行为一致

func (*ShardedMap[K, V]) CompareAndSwap ¶

func (m *ShardedMap[K, V]) CompareAndSwap(key K, old, new V) bool

CompareAndSwap 原子比较并交换 仅当 key 存在且当前值等于 old 时，才替换为 new 注意：V 为非可比较类型（如 slice/map）时会 panic，与 sync.Map 行为一致

func (*ShardedMap[K, V]) Count ¶

func (m *ShardedMap[K, V]) Count(filter func(K, V) bool) int

Count 返回满足条件的元素数量 filter 为 nil 时等价于 Len

func (*ShardedMap[K, V]) Delete ¶

func (m *ShardedMap[K, V]) Delete(key K)

Delete 删除 key

func (*ShardedMap[K, V]) Has ¶

func (m *ShardedMap[K, V]) Has(key K) bool

Has 检查 key 是否存在

func (*ShardedMap[K, V]) Keys ¶

func (m *ShardedMap[K, V]) Keys() []K

Keys 返回所有 key 的切片

func (*ShardedMap[K, V]) Len ¶

func (m *ShardedMap[K, V]) Len() int

Len 返回元素总数（原子读取，零锁开销）

func (*ShardedMap[K, V]) Load ¶

func (m *ShardedMap[K, V]) Load(key K) (V, bool)

Load 加载 key 的 value 返回：(value, exists)

func (*ShardedMap[K, V]) LoadAndDelete ¶

func (m *ShardedMap[K, V]) LoadAndDelete(key K) (V, bool)

LoadAndDelete 加载并删除 返回：(value, exists)

func (*ShardedMap[K, V]) LoadOrStore ¶

func (m *ShardedMap[K, V]) LoadOrStore(key K, value V) (V, bool)

LoadOrStore 加载或存储 如果 key 存在返回 (existing, true)，否则存储 value 并返回 (value, false)

func (*ShardedMap[K, V]) Range ¶

func (m *ShardedMap[K, V]) Range(fn func(key K, value V) bool)

Range 遍历所有键值对 遍历期间每个 shard 持有读锁（分片粒度，不影响其他 shard 写入） fn 返回 false 时停止遍历

func (*ShardedMap[K, V]) RangeParallel ¶

func (m *ShardedMap[K, V]) RangeParallel(workerNum int, fn func(key K, value V))

RangeParallel 并行遍历所有键值对（百万级数据优化）

将 shards 分配到 workerNum 个 goroutine 并行遍历，每个 goroutine 独立持有所分配 shard 的读锁。 不同 goroutine 处理不同 shard，读锁无竞争；同一 shard 仅被一个 goroutine 读取，零锁争用。

对比 Range（串行遍历）：百万级数据提速数倍，不需要 Keys() 那样的中间切片

实测（Apple M4，10 核，一百万个 int 键值对，64 shard）：

回调内容                        Range        RangeParallel
本地累加（无争用）               5.91 ms      1.56 ms      快 3.8 倍
atomic.Int64.Add（共享计数器）    6.17 ms      44.2 ms      慢 7.2 倍

**fn 里的争用会把并行的收益吃光并倒转**：十个 goroutine 抢同一个原子计数器时， 缓存行在核间来回弹跳的代价远超并行省下的时间，比串行还慢七倍 想累加就让每个 goroutine 写各自的槽位（例如按 key 分桶），遍历结束再合并； 往同一个 channel 发、抢同一把锁、累加同一个计数器都属于这一类 换句话说「提速数倍」的前提是 fn 之间互不干扰，而这一条比并行本身更决定结果

[EN] Several times faster than Range on million-scale data, with no intermediate slice. Measured on an Apple M4 with 10 cores and one million int pairs across 64 shards: a contention-free callback runs 5.91 ms serially against 1.56 ms in parallel, a 3.8x gain, while a callback adding to one shared atomic.Int64 runs 6.17 ms serially against 44.2 ms in parallel — 7.2x *slower*. Contention inside fn does not merely erase the gain, it inverts it: ten goroutines bouncing one cache line between cores cost far more than parallelism saves. Accumulate into per-goroutine slots and merge afterwards; sending to one channel, taking one lock or incrementing one counter all fall into the same trap. The promised speedup is conditional on fn not interfering with itself, and that condition matters more than the parallelism

注意：

• fn 无返回值（始终遍历全部），不支持提前终止——并行场景下"全局停止"语义不明确
• fn 在持有 shard 读锁时执行，应为非阻塞操作（TrySend 等非阻塞操作可安全调用）
• workerNum <= 0 时使用 GOMAXPROCS；shard 数量少于 worker 时退化为按 shard 数并行

func (*ShardedMap[K, V]) Store ¶

func (m *ShardedMap[K, V]) Store(key K, value V)

Store 存储 key→value

func (*ShardedMap[K, V]) StoreBatch ¶

func (m *ShardedMap[K, V]) StoreBatch(items map[K]V)

StoreBatch 批量存储，按 shard 分组减少锁获取次数 相比逐个 Store，N 个元素分散到 S 个 shard 时锁获取次数从 N 降到 S

分组用计数排序而不是 map 嵌 map

原先为每个 shard 建一个 map[K]V 来装它的那批键，一千个元素分到 64 个 shard 要分配 451 次；而 shard 下标本就是 [0, shardCount) 的小整数， 计数排序两趟即可把元素按 shard 排好，总共三次分配

为什么保留分组而不是干脆逐项 Store：量过两种写法，结论随 shard 数翻转—— 64 个 shard 时逐项更快（争用小，省下的是建表开销）， 但 8 个 shard 并发写时逐项反而慢一倍（十个 goroutine 抢八把锁，争用占主导，一千次加锁远贵于一次） 分组的价值在争用大的那一侧，去掉分配之后两侧都不亏

[EN] Grouping is done with a counting sort rather than a map of maps. The previous version built one map[K]V per shard, which cost 451 allocations for a thousand elements across 64 shards, even though a shard index is just a small integer in [0, shardCount) that two passes can order by. Grouping itself is kept rather than falling back to per-item Store because the two were measured and the winner flips with the shard count: per-item is faster with 64 shards, where contention is low and the saving is the bookkeeping, but half as fast with 8 shards under concurrent writers, where ten goroutines contend for eight locks and a thousand acquisitions cost far more than one. Grouping earns its keep on the contended side, and without the allocations neither side loses StoreBatch 批量写入

元素较多时按 shard 分组，每个 shard 只加锁一次，把加锁次数从 len(items) 降到 最多 shardCount 次分组用计数排序，两趟线性扫描，不做比较排序

分组并非总是更快，阈值取 shardCount

分组省下来的是「多个元素落进同一个 shard」时的重复加锁，所以只有 len(items) 明显大于 shardCount 时才有的省元素更少时多数 shard 只摊到一个 元素，省不下任何一次加锁，却要先付三次切片分配（entries/counts/sorted） 实测（Apple M4，10 并发，ns/op，越小越好）：

元素数   8 分片 逐项/分组     64 分片 逐项/分组
8        426 / 375           290 / 396
16       917 / 549           539 / 632
32       1829 / 848          913 / 1115
64       4029 / 1498         1897 / 1747   ← 64 分片的交点
512      34825 / 11060       15759 / 10993

两种分片数的交点都落在 len(items) ≈ shardCount 上，与上面的机理一致 无争用的单协程场景加锁只是几纳秒的原子操作，分组要到约 512 项才扳平， 因此这个阈值偏向争用场景——分片 map 本来就是为争用而存在的

[EN] Bulk store. With enough items it groups them by shard with a counting sort so each shard is locked once, cutting lock acquisitions from len(items) to at most shardCount.

Grouping is not always faster, and the threshold is shardCount. What grouping saves is the repeated locking when several items land in the same shard, so it only pays off once len(items) clearly exceeds shardCount; below that most shards hold a single item, no lock acquisition is saved, and three slice allocations are spent for nothing. Measured on an Apple M4 at 10-way parallelism, both shard counts cross over at len(items) ~ shardCount. With no contention at all grouping needs about 512 items to break even, so this threshold favours the contended case — which is what a sharded map is for

func (*ShardedMap[K, V]) Swap ¶

func (m *ShardedMap[K, V]) Swap(key K, value V) (V, bool)

Swap 替换 key 的值并返回之前的值 如果 key 不存在，存储新值并返回零值和 false

func (*ShardedMap[K, V]) Values ¶

func (m *ShardedMap[K, V]) Values() []V

Values 返回所有 value 的切片

func (*ShardedMap[K, V]) WithShardLock ¶

func (m *ShardedMap[K, V]) WithShardLock(key K, fn func(shardData map[K]V))

WithShardLock 在 key 对应的 shard 锁内执行操作（写锁） 用于需要在同一 shard 内原子操作多个 map 的场景

func (*ShardedMap[K, V]) WithShardRLock ¶

func (m *ShardedMap[K, V]) WithShardRLock(key K, fn func(shardData map[K]V))

WithShardRLock 在 key 对应的 shard 读锁内执行操作（读锁）

func (*StateMachine[S]) AllowTransition ¶

func (sm *StateMachine[S]) AllowTransition(from, to S)

AllowTransition 允许从状态 from 转换到状态 to 可以多次调用以配置状态转换规则

示例:

sm.AllowTransition("pending", "approved")
sm.AllowTransition("pending", "rejected")

func (*StateMachine[S]) AllowTransitions ¶

func (sm *StateMachine[S]) AllowTransitions(from S, toStates ...S)

AllowTransitions 批量允许状态转换 参数 from 为起始状态, toStates 为可转换到的所有目标状态

示例:

sm.AllowTransitions("pending", "approved", "rejected", "cancelled")

func (*StateMachine[S]) CanTransitionTo ¶

func (sm *StateMachine[S]) CanTransitionTo(to S) bool

CanTransitionTo 检查是否可以转换到目标状态 线程安全

示例:

if sm.CanTransitionTo("approved") {
    可以转换
}

func (*StateMachine[S]) ClearCallbacks ¶

func (sm *StateMachine[S]) ClearCallbacks()

ClearCallbacks 清除所有回调函数

func (*StateMachine[S]) ClearHistory ¶

func (sm *StateMachine[S]) ClearHistory()

ClearHistory 清除历史记录

func (*StateMachine[S]) CurrentState ¶

func (sm *StateMachine[S]) CurrentState() S

CurrentState 获取当前状态 线程安全

func (*StateMachine[S]) DisallowTransition ¶

func (sm *StateMachine[S]) DisallowTransition(from, to S)

DisallowTransition 禁止从状态 from 转换到状态 to

func (*StateMachine[S]) GetAllowedTransitions ¶

func (sm *StateMachine[S]) GetAllowedTransitions() []S

GetAllowedTransitions 获取当前状态允许转换到的所有状态

func (*StateMachine[S]) GetHistory ¶

func (sm *StateMachine[S]) GetHistory() []StateTransition[S]

GetHistory 获取状态转换历史记录 返回所有历史记录的副本,按时间顺序排列(最早的在前)

示例:

history := sm.GetHistory()
for _, trans := range history {
    fmt.Printf("%s: %v -> %v\n", trans.Timestamp.Format(time.RFC3339), trans.From, trans.To)
}

func (*StateMachine[S]) GetHistoryCount ¶

func (sm *StateMachine[S]) GetHistoryCount() int

GetHistoryCount 获取历史记录数量

func (*StateMachine[S]) GetHistoryString ¶

func (sm *StateMachine[S]) GetHistoryString() string

GetHistoryString 获取完整执行流程的字符串表示 使用创建状态机时配置的时间格式 返回格式化后的历史记录字符串,每行一条记录

示例:

historyStr := sm.GetHistoryString()
fmt.Println(historyStr)

func (*StateMachine[S]) GetLastTransition ¶

func (sm *StateMachine[S]) GetLastTransition() (StateTransition[S], bool)

GetLastTransition 获取最后一次状态转换记录 如果没有历史记录，返回零值和 false

func (*StateMachine[S]) GetLastTransitionDuration ¶

func (sm *StateMachine[S]) GetLastTransitionDuration() time.Duration

GetLastTransitionDuration 获取上次状态转换的持续时间（在上一个状态停留的时间） 如果没有历史记录，返回 0

func (*StateMachine[S]) GetLastTransitionTime ¶

func (sm *StateMachine[S]) GetLastTransitionTime() time.Time

GetLastTransitionTime 获取上次状态转换的时间

func (*StateMachine[S]) GetRecentHistory ¶

func (sm *StateMachine[S]) GetRecentHistory(n int) []StateTransition[S]

GetRecentHistory 获取最近 n 条状态转换历史记录 如果 n 大于实际历史记录数量,返回所有历史记录

示例:

recentHistory := sm.GetRecentHistory(5) // 获取最近5条记录

func (*StateMachine[S]) GetRecentHistoryString ¶

func (sm *StateMachine[S]) GetRecentHistoryString(n int) string

GetRecentHistoryString 获取最近 n 条执行流程的字符串表示 n 为记录数量，使用创建状态机时配置的时间格式

示例:

historyStr := sm.GetRecentHistoryString(5)

func (*StateMachine[S]) GetTotalDuration ¶

func (sm *StateMachine[S]) GetTotalDuration() time.Duration

GetTotalDuration 获取从状态机创建到当前的总时长

func (*StateMachine[S]) IsTrackingHistory ¶

func (sm *StateMachine[S]) IsTrackingHistory() bool

IsTrackingHistory 检查是否正在追踪历史记录

func (*StateMachine[S]) MustTransitionTo ¶

func (sm *StateMachine[S]) MustTransitionTo(to S)

MustTransitionTo 强制转换到目标状态,如果失败则 panic

func (*StateMachine[S]) OnEnter ¶

func (sm *StateMachine[S]) OnEnter(state S, handler func(from S))

OnEnter 注册进入某个状态时的回调 参数 from 为转换前的状态

示例:

sm.OnEnter("connected", func(from string) {
    log.Printf("Entered connected state from %s", from)
})

func (*StateMachine[S]) OnExit ¶

func (sm *StateMachine[S]) OnExit(state S, handler func(to S))

OnExit 注册离开某个状态时的回调 参数 to 为转换后的目标状态

示例:

sm.OnExit("connected", func(to string) {
    log.Printf("Exiting connected state to %s", to)
})

func (*StateMachine[S]) OnTransition ¶

func (sm *StateMachine[S]) OnTransition(handler func(from, to S))

OnTransition 注册状态转换回调 每次状态转换时都会调用所有已注册的回调函数

示例:

sm.OnTransition(func(from, to string) {
    log.Printf("State changed: %s -> %s", from, to)
})

func (*StateMachine[S]) Reset ¶

func (sm *StateMachine[S]) Reset(initialState S)

Reset 重置状态机到初始状态

func (*StateMachine[S]) TransitionTo ¶

func (sm *StateMachine[S]) TransitionTo(to S) error

TransitionTo 转换到目标状态 如果转换不被允许,返回错误 转换成功时会触发所有已注册的回调函数

示例:

if err := sm.TransitionTo("approved"); err != nil {
    log.Printf("Transition failed: %v", err)
}

func (*Task[T, R, U]) AddDependency ¶

func (tk *Task[T, R, U]) AddDependency(dep *Task[T, R, U]) *Task[T, R, U]

AddDependency 添加依赖关系

func (*Task[T, R, U]) GetCallbackDuration ¶

func (tk *Task[T, R, U]) GetCallbackDuration() time.Duration

GetCallbackDuration 获取回调运行时间

func (*Task[T, R, U]) GetCallbackError ¶

func (tk *Task[T, R, U]) GetCallbackError() error

GetCallbackError 获取回调错误

func (*Task[T, R, U]) GetCallbackResult ¶

func (tk *Task[T, R, U]) GetCallbackResult() U

GetCallbackResult 获取回调结果

func (*Task[T, R, U]) GetCallbackState ¶

func (tk *Task[T, R, U]) GetCallbackState() TaskState

GetCallbackState 获取回调状态

func (*Task[T, R, U]) GetDependExecutionMode ¶

func (tk *Task[T, R, U]) GetDependExecutionMode() ExecutionMode

GetDependExecutionMode 获取依赖任务的执行模式

func (*Task[T, R, U]) GetDependencyStates ¶

func (tk *Task[T, R, U]) GetDependencyStates() map[string]TaskState

GetDependencyStates 获取所有依赖任务的状态

func (*Task[T, R, U]) GetDepends ¶

func (tk *Task[T, R, U]) GetDepends() []*Task[T, R, U]

GetDepends 获取当前任务的所有依赖任务

func (*Task[T, R, U]) GetError ¶

func (tk *Task[T, R, U]) GetError() error

GetError 获取任务执行错误

func (*Task[T, R, U]) GetFnDuration ¶

func (tk *Task[T, R, U]) GetFnDuration() time.Duration

GetFnDuration 获取主任务运行时间

func (*Task[T, R, U]) GetInput ¶

func (tk *Task[T, R, U]) GetInput() T

GetInput 获取任务输入

func (*Task[T, R, U]) GetMaxRetries ¶

func (tk *Task[T, R, U]) GetMaxRetries() int32

GetMaxRetries 获取最大重试次数

func (*Task[T, R, U]) GetName ¶

func (tk *Task[T, R, U]) GetName() string

GetName 获取任务名称

func (*Task[T, R, U]) GetResult ¶

func (tk *Task[T, R, U]) GetResult() R

GetResult 获取任务结果

func (*Task[T, R, U]) GetRetryCount ¶

func (tk *Task[T, R, U]) GetRetryCount() int32

GetRetryCount 获取任务失败后重试次数

func (*Task[T, R, U]) GetState ¶

func (tk *Task[T, R, U]) GetState() TaskState

GetState 获取任务状态

func (*Task[T, R, U]) GetTaskHistory ¶

func (tk *Task[T, R, U]) GetTaskHistory(taskName string) []TaskHistory

GetTaskHistory 获取任务执行历史

func (*Task[T, R, U]) GetTaskType ¶

func (tk *Task[T, R, U]) GetTaskType() TaskType

GetTaskType 获取任务类型

func (*Task[T, R, U]) GetTimestamp ¶

func (tk *Task[T, R, U]) GetTimestamp() int64

GetTimestamp 获取任务开始时间戳（纳秒）

func (*Task[T, R, U]) GetTotalDuration ¶

func (tk *Task[T, R, U]) GetTotalDuration() time.Duration

GetTotalDuration 获取任务的总耗时

func (*Task[T, R, U]) SetDependExecutionMode ¶

func (tk *Task[T, R, U]) SetDependExecutionMode(executionMode ExecutionMode) *Task[T, R, U]

SetDependExecutionMode 设置依赖任务的执行模式，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetFailureCallback ¶

func (tk *Task[T, R, U]) SetFailureCallback(callback TaskCallbackFunc[R, U]) *Task[T, R, U]

SetFailureCallback 设置任务失败后的回调函数，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetMaxHistorySize ¶

func (tk *Task[T, R, U]) SetMaxHistorySize(maxHistorySize int) *Task[T, R, U]

SetMaxHistorySize 设置最大历史记录行到 TaskManager

func (*Task[T, R, U]) SetMaxRetries ¶

func (tk *Task[T, R, U]) SetMaxRetries(count int32) *Task[T, R, U]

SetMaxRetries 设置最大重试次数，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetPriority ¶

func (tk *Task[T, R, U]) SetPriority(priority int) *Task[T, R, U]

SetPriority 设置任务优先级，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetRetryInterval ¶

func (tk *Task[T, R, U]) SetRetryInterval(retryInterval time.Duration) *Task[T, R, U]

SetRetryInterval 设置任务失败后重试间隔时间，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetSuccessCallback ¶

func (tk *Task[T, R, U]) SetSuccessCallback(callback TaskCallbackFunc[R, U]) *Task[T, R, U]

SetSuccessCallback 设置任务成功后的回调函数，并返回当前任务以支持链式调用

func (*Task[T, R, U]) SetTimeout ¶

func (tk *Task[T, R, U]) SetTimeout(timeout time.Duration) *Task[T, R, U]

SetTimeout 设置任务超时时间，并返回当前任务以支持链式调用

func (*TaskHistory) GetCallbackDuration ¶

func (th *TaskHistory) GetCallbackDuration() time.Duration

GetCallbackDuration 获取回调函数的执行时间

func (*TaskHistory) GetError ¶

func (th *TaskHistory) GetError() error

GetError 获取任务执行过程中发生的错误

func (*TaskHistory) GetFnDuration ¶

func (th *TaskHistory) GetFnDuration() time.Duration

GetFnDuration 获取任务函数的执行时间

func (*TaskHistory) GetResult ¶

func (th *TaskHistory) GetResult() interface{}

GetResult 获取任务执行的结果

func (*TaskHistory) GetState ¶

func (th *TaskHistory) GetState() TaskState

GetState 获取任务的状态

func (*TaskHistory) GetTaskType ¶

func (th *TaskHistory) GetTaskType() TaskType

GetTaskType 获取任务类型

func (*TaskHistory) GetTimestamp ¶

func (th *TaskHistory) GetTimestamp() int64

GetTimestamp 获取任务执行的时间戳

func (*TaskManager[T, R, U]) AddTask ¶

func (tm *TaskManager[T, R, U]) AddTask(task *Task[T, R, U]) *TaskManager[T, R, U]

AddTask 添加一个任务到 TaskManager

func (*TaskManager[T, R, U]) Cancel ¶

func (tm *TaskManager[T, R, U]) Cancel(taskName string)

Cancel 取消某个任务

func (*TaskManager[T, R, U]) CancelAll ¶

func (tm *TaskManager[T, R, U]) CancelAll()

CancelAll 取消所有任务

func (*TaskManager[T, R, U]) GetTask ¶

func (tm *TaskManager[T, R, U]) GetTask(name string) (*Task[T, R, U], bool)

GetTask 获取指定名称的任务（线程安全）

func (*TaskManager[T, R, U]) GetTasks ¶

func (tm *TaskManager[T, R, U]) GetTasks() map[string]*Task[T, R, U]

GetTasks 获取所有任务的快照副本（避免外部并发修改内部 map）

func (*TaskManager[T, R, U]) Run ¶

func (tm *TaskManager[T, R, U]) Run() error

Run 执行所有任务

func (*TaskManager[T, R, U]) SetTrunDown ¶

func (tm *TaskManager[T, R, U]) SetTrunDown(fn TaskTrunFunc[R]) *TaskManager[T, R, U]

SetTrunDown 设置关闭时执行的函数

func (*TaskManager[T, R, U]) SetTrunUp ¶

func (tm *TaskManager[T, R, U]) SetTrunUp(fn TaskTrunFunc[R]) *TaskManager[T, R, U]

SetTrunUp 设置启动时执行的函数

func (*TaskManager[T, R, U]) TrunDown ¶

func (tm *TaskManager[T, R, U]) TrunDown() (result R, err error)

TrunDown 关闭任务管理器并返回结果和错误

func (*TaskManager[T, R, U]) TrunUp ¶

func (tm *TaskManager[T, R, U]) TrunUp() (result R, err error)

TrunUp 启动任务管理器并返回结果和错误

func (*Uint32) Add ¶

func (u32 *Uint32) Add(i uint32) uint32

Add 原子地增加值

func (*Uint32) CAS ¶

func (u32 *Uint32) CAS(old, new uint32) bool

CAS 原子地比较并交换值

func (*Uint32) Load ¶

func (u32 *Uint32) Load() uint32

Load 原子地加载值

func (*Uint32) Store ¶

func (u32 *Uint32) Store(i uint32)

Store 原子地存储值

func (*Uint32) String ¶

func (u32 *Uint32) String() string

String 返回值的字符串表示

func (*Uint32) Sub ¶

func (u32 *Uint32) Sub(i uint32) uint32

Sub 原子地减少值

func (*Uint32) Swap ¶

func (u32 *Uint32) Swap(i uint32) uint32

Swap 原子地交换值

func (*Uint64) Add ¶

func (u64 *Uint64) Add(i uint64) uint64

Add 原子地增加值

func (*Uint64) CAS ¶

func (u64 *Uint64) CAS(old, new uint64) bool

CAS 原子地比较并交换值

func (*Uint64) Load ¶

func (u64 *Uint64) Load() uint64

Load 原子地加载值

func (*Uint64) Store ¶

func (u64 *Uint64) Store(i uint64)

Store 原子地存储值

func (*Uint64) String ¶

func (u64 *Uint64) String() string

String 返回值的字符串表示

func (*Uint64) Sub ¶

func (u64 *Uint64) Sub(i uint64) uint64

Sub 原子地减少值

func (*Uint64) Swap ¶

func (u64 *Uint64) Swap(i uint64) uint64

Swap 原子地交换值

func (*Uintptr) CAS ¶

func (ptr *Uintptr) CAS(old, new uintptr) bool

CAS 原子地比较并交换值

func (*Uintptr) Load ¶

func (ptr *Uintptr) Load() uintptr

Load 原子地加载值

func (*Uintptr) Store ¶

func (ptr *Uintptr) Store(i uintptr)

Store 原子地存储值

func (*Uintptr) String ¶

func (ptr *Uintptr) String() string

String 返回值的字符串表示

func (*Uintptr) Swap ¶

func (ptr *Uintptr) Swap(i uintptr) uintptr

Swap 原子地交换值

func (*WorkerPool) Close ¶

func (p *WorkerPool) Close() error

Close 关闭 Worker 池，等待所有任务完成 调用此方法后，所有新的提交都会返回 ErrClosed

注意：不关闭 queue channel，避免 Submit 在检查 closed 标志后 仍向已关闭 channel 发送数据导致 panic。worker 通过 ctx.Done() 退出

示例:

pool := NewWorkerPool(20, 100)
提交任务 ...
pool.Close()  // 等待所有任务完成后返回

func (*WorkerPool) GetActiveWorkerCount ¶

func (p *WorkerPool) GetActiveWorkerCount() int

GetActiveWorkerCount 获取当前存活的 worker 数量

func (*WorkerPool) GetQueueSize ¶

func (p *WorkerPool) GetQueueSize() int

GetQueueSize 获取队列中待处理任务数

func (*WorkerPool) GetWorkerCount ¶

func (p *WorkerPool) GetWorkerCount() int

GetWorkerCount 获取 worker 数量上限（弹性伸缩的并发安全帽）

func (*WorkerPool) IsClosed ¶

func (p *WorkerPool) IsClosed() bool

IsClosed 检查 Worker 池是否已关闭

func (*WorkerPool) Submit ¶

func (p *WorkerPool) Submit(ctx context.Context, task WorkerTask) error

Submit 提交任务到队列 如果队列满，会阻塞直到有空位或 context 取消

参数:

• ctx: 上下文，用于超时和取消控制
• task: 要执行的任务

返回:

• nil: 任务成功提交
• ErrClosed: Worker 池已关闭
• context.Err(): context 被取消或超时

示例:

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

err := pool.Submit(ctx, func() {
   处理任务
})
if err != nil {
    log.Errorf("submit failed: %v", err)
}

func (*WorkerPool) SubmitNonBlocking ¶

func (p *WorkerPool) SubmitNonBlocking(task WorkerTask) error

SubmitNonBlocking 非阻塞提交任务 如果队列满，返回错误而不是阻塞

参数:

• task: 要执行的任务

返回:

• nil: 任务成功提交
• ErrClosed: Worker 池已关闭
• ErrQueueFull: 队列已满

示例:

err := pool.SubmitNonBlocking(func() {
   处理任务
})
if err == ErrQueueFull {
    log.Warn("queue is full, task dropped")
}

func (*WorkerPool) Wait ¶

func (p *WorkerPool) Wait()

Wait 等待所有已提交的任务完成 不关闭 pool，可以继续提交新任务

示例:

pool := NewWorkerPool(20, 100)
提交任务 ...
pool.Wait()  // 等待所有任务完成
继续提交任务 ...