syncx

package
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 8, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Overview

Package syncx 提供一组便捷的并发原语(泛型),补齐标准库与 pkg/xgo 之外常被手搓、 又容易写错的模式:

  • Map / ForEach:对一批元素并发处理,带并发上限 + 错误聚合(首个出错即取消其余);
  • SingleFlight:并发相同 key 的调用去重合并,防缓存击穿/惊群;
  • Batcher:攒够 N 条或到时间就 flush,批量写库/推送/调用;
  • Debounce / Throttle:事件去抖 / 限频;
  • Future / Async:异步跑一个函数,稍后 Await 取结果。

仅依赖标准库与 golang.org/x/sync。

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Debounce

func Debounce(d time.Duration, fn func()) (call func(), cancel func())

Debounce 返回一个去抖函数 call:每次调用都会把 fn 的执行推迟 d;只有在 d 内不再被调用时 fn 才触发 (即"最后一次调用后静默 d 才执行")。cancel 取消尚未触发的待执行。适合配置热更防抖、搜索联想等。 fn 在独立的 time.AfterFunc goroutine 里执行。

func ForEach

func ForEach[T any](ctx context.Context, items []T, limit int, fn func(context.Context, T) error) error

ForEach 对 items 并发执行 fn(不收集结果),最多 limit 个并发(limit<=0 不限)。 任一出错取消其余并返回首个错误。

func Map

func Map[T, R any](ctx context.Context, items []T, limit int, fn func(context.Context, T) (R, error)) ([]R, error)

Map 对 items 并发执行 fn,最多 limit 个并发(limit<=0 表示不限),按输入顺序返回结果。 任一 fn 出错则取消其余(通过传入 fn 的 ctx)并返回首个错误。

func Throttle

func Throttle(d time.Duration, fn func()) func()

Throttle 返回一个限频函数:每 d 内最多触发一次 fn(前沿触发——窗口内首次调用立即执行, 其余忽略)。适合按钮防连点、高频事件降频。fn 同步执行。

Types

type Batcher

type Batcher[T any] struct {
	// contains filtered or unexported fields
}

Batcher 把逐条 Add 的元素攒成批,达到 maxSize 条或距上批首条满 maxWait 时,调用 flush 处理一批。 适合批量写库 / 批量推送 / 批量调用。单后台 goroutine 收集,Add 并发安全。零值不可用,用 NewBatcher 构造。

func NewBatcher

func NewBatcher[T any](maxSize int, maxWait time.Duration, flush func([]T)) *Batcher[T]

NewBatcher 创建批处理器。maxSize>0 为批容量上限,maxWait>0 为一批的最大等待时长(到点即使不满也 flush)。 flush 在后台 goroutine 串行调用,收到的 slice 仅在本次调用内有效(下一批会复用底层数组前不会重叠, 但调用方若需长期持有请自行拷贝)。

func (*Batcher[T]) Add

func (b *Batcher[T]) Add(item T)

Add 加入一条元素。Close 之后调用会被丢弃(不 panic)。

func (*Batcher[T]) Close

func (b *Batcher[T]) Close()

Close 停止批处理器,flush 掉剩余元素后返回。幂等。

type Future

type Future[T any] struct {
	// contains filtered or unexported fields
}

Future 是一个异步计算的结果句柄。用 Async 发起,Await 取结果。

func Async

func Async[T any](fn func() (T, error)) *Future[T]

Async 在新 goroutine 里执行 fn,返回 Future;fn 内的 panic 会被捕获转成错误。

func AsyncCtx added in v0.3.0

func AsyncCtx[T any](ctx context.Context, fn func(ctx context.Context) (T, error)) *Future[T]

AsyncCtx 在新 goroutine 里执行 fn 并传入 ctx;ctx 取消时 fn 可主动退出,避免后台泄漏。

func (*Future[T]) Await

func (f *Future[T]) Await(ctx context.Context) (T, error)

Await 等待结果,或在 ctx 取消时返回 ctx 错误(此时后台计算仍会继续到自然结束)。

func (*Future[T]) Done

func (f *Future[T]) Done() <-chan struct{}

Done 返回完成信号 channel,便于在 select 中组合。

type Group

type Group[V any] struct {
	// contains filtered or unexported fields
}

Group 对相同 key 的并发调用去重合并:同一时刻多个 goroutine Do 同一 key,只有一个真正执行 fn,其余共享其结果——防缓存击穿/惊群。零值可用。V 是结果类型。

func (*Group[V]) Do

func (g *Group[V]) Do(key string, fn func() (V, error)) (V, error, bool)

Do 执行(或复用正在执行的)key 对应的 fn,返回结果、错误,以及本次结果是否被多个调用共享。

func (*Group[V]) Forget

func (g *Group[V]) Forget(key string)

Forget 让 key 的下一次 Do 重新执行(不再复用在途结果)。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL