Documentation
¶
Overview ¶
Package syncx 并发编排原语（stdlib-only）：在飞闸门（按键分档位 / 全局单一 配额，取用分排队与即拒两档）+ 按序扇出收集。 收编自消费方手写形态——LLM 按模型名限在飞（429 风暴治理）、「这个进程总共 最多 N 个在飞」的手搓信号量，与「按 index 写结果免锁」的并行扇出样板。
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Gate ¶ added in v0.19.0
type Gate struct {
// contains filtered or unexported fields
}
Gate 是**全局在飞上限闸门**：整个实例最多 cap 个在飞，排队者等 ctx 取消 即返回 ctx.Err()。与 KeyedGate 的区别：Gate 是单一全局配额（不分 key）， KeyedGate 按 key 分档位（按模型名/租户名分别限流）。
两种取用语义并存、命名区分：
- Acquire(ctx)：满额时**排队**等名额；
- TryAcquire()：满额时**立即拒**（不等待、不占名额）——重活的并发上限 用满就该立刻拒，让调用方回 429 / 503，而不是把请求堆在队列里。
为什么需要这一档：现有原语都不覆盖"这个进程总共最多 N 个在飞"这类闸门。
- KeyedGate 按 key 隔离，用固定 key 硬凑可行，但那是把"多档位闸门"原语 当"单档位闸门"用，语义与命名都不对；
- gopool.Pool 是弹性 worker 池，**MaxIdleWorkers 是常驻 worker 上限， 不是并发上限**——任务队列满时回退为独立 goroutine 直跑，任务永不因 并发数被拒（实测 MaxIdleWorkers=2 时实际并发仍是 10）；它与"硬配额" 的语义正好相反；
- ratelimit.KeyedLimiter 是速率型（令牌桶），不是并发闸门。
适用场景：保护下游 DB / 控制子进程数 / 限制外部 API 并发。
并发安全；release 幂等（可重复调用，只归还一次）。
func (*Gate) Acquire ¶ added in v0.19.0
Acquire 取一个在飞名额；ctx 取消（排队等待中）返回 ctx.Err()。 返回的 release 归还名额，幂等可重复调用。
ctx 为 nil 时按 context.Background() 处理（永不取消）。
func (*Gate) TryAcquire ¶ added in v0.20.0
TryAcquire 尝试取一个在飞名额，**不等待**：名额已满时**立即**返回 ok=false。
与 Acquire 的区别是**排队 vs 即拒**，两者并存、命名区分：
- Acquire(ctx)：满额时**排队**等名额，等 ctx 取消才返回 ctx.Err()；
- TryAcquire()：满额时**立刻拒**，一个名都不占。
什么时候要即拒：Acquire 的排队语义会把「并发被占满」换成「请求堆在队列里」—— 那是另一种 DoS 形态。重活（图片全量解码、ffmpeg 子进程、模型推理）的并发 上限用满就该立刻拒，让调用方回 429 / 503，而不是让队列越堆越长、把内存 或子进程数推到别处爆掉。
ok=false 只有一个含义：**名额已满**（非阻塞取名额没有别的失败原因）， 调用方可据此返回 429 / 503，而不是泛化成 500。
返回的 release 归还名额，幂等可重复调用（与 Acquire 一致）。 cap<=0（不限）时恒返回 ok=true——不限额的闸门没有「满」这回事。
用法:
release, ok := gate.TryAcquire()
if !ok {
return nil, ErrTooManyInFlight // 429 / 503
}
defer release()
type KeyedGate ¶
type KeyedGate struct {
// contains filtered or unexported fields
}
KeyedGate 按键在飞上限闸门（并发安全；channel 令牌先到先得；键集由调用方 保证有界——如按模型名/租户名，键量级恒小不回收）。
按 key 分档位（"每个模型最多 N 个在飞"）。要的是**单一全局配额** （"整个进程总共最多 N 个在飞"）用 Gate——两者语义不同，不要用固定 key 硬凑。
func NewKeyedGate ¶
NewKeyedGate 构造；cap<=0 = 不限（Acquire 直通零开销）。
func (*KeyedGate) Acquire ¶
Acquire 取一个键下在飞名额；ctx 取消（排队等待中）返回 ctx.Err()。 返回的 release 归还名额，幂等可重复调用。
幂等标记用 atomic.Bool + CompareAndSwap（同包 Gate 的做法）：裸 bool 在 两个 goroutine 并发调用同一个 release 时是 data race，且两者都可能读到 "未归还" 而各执行一次 <-ch —— 名额被归还两次，该 key 的在飞数突破 cap 上限（cap=1 时可同时有 2 个在飞）。按 key 分档位的闸门正是最容易被并发 释放的场景（超时清理路径与正常返回路径同时释放同一个 key）。
ctx 为 nil 时按 context.Background() 处理（永不取消），与 Gate 一致。
func (*KeyedGate) TryAcquire ¶ added in v0.21.0
TryAcquire 尝试取一个键下在飞名额，**不等待**：该 key 档位已满时**立即**返回 ok=false（不排队、不占名额）。
与 Acquire 的区别是**排队 vs 即拒**，两者并存、命名区分（同包 Gate 的同款分档）：
- Acquire(ctx, key)：该 key 满额时**排队**等名额，等 ctx 取消才返回 ctx.Err()；
- TryAcquire(key)：该 key 满额时**立刻拒**，一个名都不占。
什么时候要即拒：防重入场景的常见形态——同一个 key 正在处理中就**立刻拒绝** （回 409/429），而不是排队等它做完。此前消费方只能手搓 map[K]*sync.Mutex + TryLock + sync.Once 兜底幂等归还：每仓各写一遍，once 包装又容易漏（漏了就是 双重解锁 panic）。KeyedGate.Acquire 是排队语义，套不上——要「忙即拒」就得上 调用方自己再套一层 goroutine + select，绕开了原语的语义。
ok=false 只有一个含义：**该 key 档位已满**（非阻塞取名额没有别的失败原因）， 调用方可据此返回 409 / 429，而不是泛化成 500。
返回的 release 归还名额，幂等可重复调用（与 Acquire 一致，atomic CAS 实现）。 cap<=0（不限）时恒返回 ok=true——不限额的闸门没有「满」这回事。
用法:
release, ok := gate.TryAcquire("account:42:sync")
if !ok {
return errBusy // 409 / 429
}
defer release()
type Slot ¶ added in v0.5.0
type Slot[T any] struct { Val T // Err 正常错误，或 panic 收编的 *async.PanicError。 Err error // Panicked 历史兼容字段：panic 原始 payload。新代码改用 // errors.As(&async.PanicError) 读 Err。 // // Deprecated: v0.7.0 起由 Slot.Err 承载。 Panicked any }
Slot 是 FanInErr 的逐槽结果：Val/Err 为 fn 的正常返回；fn panic 时 Err 承载 *async.PanicError（errors.As 可辨、携带栈，#12）且 Val 为零值。 Panicked 保留原始 payload（兼容字段，Deprecated——改用 errors.As(Slot.Err)）。
func FanInErr ¶ added in v0.5.0
FanInErr 逐槽错误变体：并行执行 fn(0..n-1)（fn 可返回 error），全部完成后 按 index 序返回逐槽结果。fn 正常返回 → Val/Err 如实记录；fn panic → 该槽 Err=*async.PanicError（v0.7.0 起统一收编，errors.As 可辨、携带栈；Panicked 同写原始 payload，已 Deprecated）、Val 零值，并经 ReportPanic 上报——不再 静默（FanIn 契约保留不动，供无 error 形态的轻量路径）。n<=0 返回空切片。
func FanInLimit ¶ added in v0.12.0
FanInLimit 是 FanInErr 的受限并发变体（#23）：n 槽结果语义不变，内部固定 w = min(limit, n) 个 worker 经原子计数器拉取 index 执行——goroutine O(w)， 供「n 可达数十以上、fn 是有成本 IO（DB/远端查询）」的 fan-out 限并发， 防下游过载（收编消费方手写的 channel 信号量样板）。其余语义与 FanInErr 一致：按 index 序返回逐槽结果、panic 进该槽 Err（*async.PanicError）并 ReportPanic 上报。limit<=0 = 不设上限（等价 FanInErr 行为）。
func FanInLimitCtx ¶ added in v0.14.0
func FanInLimitCtx[T any](ctx context.Context, n, limit int, fn func(ctx context.Context, i int) (T, error)) []Slot[T]
FanInLimitCtx 是 FanInLimit 的 ctx 感知变体（#27）：fn 携带 ctx 供单项内部 使用；引擎级取消语义——worker 每轮拉取 index 前先查 ctx.Err()（worker 内 缓存，取消后零额外锁开销），命中即不再启动新工作（fn 一次都不调），当前及 剩余槽一次遍历快速排空为该错误（context.Canceled）；已在飞 item 跑完、结果 照常落槽。其余语义与 FanInLimit 一致（panic 进槽并上报、limit<=0 = 不设上限、 n<=0 = nil；ctx 为 nil 时按 context.Background() 处理）。
Source Files