flx

package module
v0.1.3 Latest Latest
Warning

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

 Go to latest Published: Mar 27, 2026 License: MIT Imports: 12 Imported by: 0

README

flx

flx 是一个以泛型 Stream[T] 为核心的流式处理与动态并发控制库。

它基于 fx 的并发语义重建，但不再保留为了兼容 go-zero fx API 而存在的包装层。整体目标很直接：

  • 用 Stream[T] 提供类型安全的流式处理
  • 保留固定并发、无限并发、动态并发和强制缩容能力
  • 用显式 context.Context 替代隐式 context option
  • 收敛旧版兼容接口，让 API 更贴近现代 Go

适用场景

  • 对一批数据做过滤、映射、分组、聚合
  • 需要在 pipeline 中控制并发度
  • 需要在运行中动态调整 worker 数量
  • 需要在缩容时取消多余 worker
  • 需要在流式处理之外复用 retry / timeout / parallel 工具

版本要求

  • Go 1.26.1+

当前模块路径是 github.com/ezra-sullivan/flx，示例代码直接使用：

import "github.com/ezra-sullivan/flx"

如果后续发布到新的模块路径，示例中的导入路径需要同步替换。

快速示例

package main

import (
	"fmt"

	"github.com/ezra-sullivan/flx"
)

func main() {
	out := flx.Map(
		flx.Values(1, 2, 3, 4, 5).
			Filter(func(v int) bool { return v%2 == 0 }),
		func(v int) int { return v * 10 },
	)

	out.ForEach(func(v int) {
		fmt.Println(v)
	})
}

输出：

20
40

API 形状

flx 采用“同类型操作保留方法，跨类型操作使用包级泛型函数”的混合设计。

例如：

  • 同类型操作：Filter、Sort、Head、Skip
  • 跨类型操作：flx.Map、flx.FlatMap、flx.MapContext
s := flx.Values(1, 2, 3).Filter(func(v int) bool { return v > 1 })
out := flx.Map(s, func(v int) string { return fmt.Sprintf("v=%d", v) })

这样设计不是为了追求风格特别，而是因为 Go 1.26 仍然不支持“额外类型参数的方法”。

核心能力

Stream 构造与中间操作
  • Values / From / FromChan / Concat
  • Filter / Buffer / Sort / Reverse
  • Head / Tail / Skip
  • Map / MapErr
  • FlatMap / FlatMapErr
  • MapContext / FlatMapContext
  • DistinctBy / GroupBy / Chunk
  • DistinctByCount / GroupByCount
  • DistinctByWindow / GroupByWindow
  • Reduce
终结与查询
  • Done / DoneErr
  • ForEach / ForEachErr
  • Parallel / ParallelErr
  • Count / Collect
  • First / Last
  • AllMatch / AnyMatch / NoneMatch
  • Max / Min
  • Err
并发与错误控制
  • WithWorkers
  • WithUnlimitedWorkers
  • WithDynamicWorkers
  • WithForcedDynamicWorkers
  • WithInterruptibleWorkers（兼容别名）
  • WithErrorStrategy
  • ErrorStrategyFailFast
  • ErrorStrategyCollect
  • ErrorStrategyLogAndContinue
独立工具函数
  • Parallel / ParallelErr / ParallelWithErrorStrategy
  • DoWithRetry / DoWithRetryCtx
  • DoWithTimeout / DoWithTimeoutCtx

内存边界

  • DistinctBy 会维护当前流的全局已见 key 集合；如果唯一 key 持续增长，内存也会持续增长。
  • GroupBy 会先缓存整条输入流，再按 key 输出分组；它不适合超大数据集或无界流。
  • 这两个 API 更适合有明确边界的批处理数据。
  • DistinctByCount / GroupByCount 采用按输入数量切分的 tumbling window，适合需要更硬内存边界的去重/分组场景。
  • DistinctByWindow / GroupByWindow 采用 processing-time tumbling window，并显式接受 context.Context；它们更适合做按时归档或周期性输出。

动态并发

优雅缩容

缩容时不打断已运行 worker，只影响后续竞争 slot 的 worker：

ctrl := flx.NewConcurrencyController(4)

out := flx.Map(
	flx.Values(1, 2, 3, 4, 5),
	func(v int) int { return v * 10 },
	flx.WithDynamicWorkers(ctrl),
)

ctrl.SetWorkers(8)
ctrl.SetWorkers(2)

out.Done()
强制缩容

缩容时取消多余 worker，要求使用 MapContext* 或 FlatMapContext*：

ctx := context.Background()
ctrl := flx.NewConcurrencyController(4)

out := flx.FlatMapContext(
	ctx,
	flx.Values("a", "b", "c"),
	func(ctx context.Context, v string, pipe chan<- string) {
		select {
		case <-ctx.Done():
			return
		default:
		}

		flx.SendContext(ctx, pipe, v+"!")
	},
	flx.WithForcedDynamicWorkers(ctrl),
)

ctrl.SetWorkers(1)
out.Done()

WithInterruptibleWorkers 仍然可用，但新代码建议统一写成 WithForcedDynamicWorkers。

MapContext / MapContextErr 的单结果发送现在也会响应 ctx.Done()；如果你在 FlatMapContext* 里自己向下游发送值，仍然建议显式使用 SendContext。

错误处理建议

默认错误策略是 ErrorStrategyFailFast。如果 worker 返回错误或 panic：

  • fail-fast：尽快取消当前操作，并在终结阶段暴露错误
  • collect：继续执行，最后合并错误
  • log-and-continue：记录日志并继续

业务代码里，如果你需要稳定的最终错误边界，优先使用会完整消费 source 的 *Err 终结操作，例如 DoneErr / CollectErr：

out := flx.MapErr(flx.Values("1", "x", "3"), strconv.Atoi)
items, err := out.CollectErr()

补充说明：

  • From 的生产函数如果 panic，会进入 stream 错误状态
  • DoneErr / CollectErr 等完整消费型 *Err 终结操作可以显式拿到这类错误
  • FirstErr / AllMatchErr / AnyMatchErr / NoneMatchErr 现在是真正短路：命中结果后立即返回，并在后台 drain 上游
  • 这四个短路 *Err API 返回的是当前错误快照；返回后才发生的 fail-fast error 不保证包含在返回值里
  • First / AllMatch / AnyMatch / NoneMatch 这类短路终结操作也遵循 fail-fast 语义；如果上游已经记录 fail-fast 错误，它们会像其他非 *Err 终结操作一样 panic

与 fx 的主要差异

  • fx.Just -> flx.Values
  • fx.Range -> flx.FromChan
  • stream.Map(...) -> flx.Map(stream, ...)
  • stream.Walk(...) -> flx.FlatMap(stream, ...)
  • stream.WalkCtx... -> flx.MapContext... / flx.FlatMapContext...
  • stream.Merge() -> stream.Collect() / stream.CollectErr()
  • fx.WithDynamicWorkersCtx -> flx.WithForcedDynamicWorkers
  • fx.SendCtx -> flx.SendContext

完整迁移表见 doc/fx-to-flx-migration.md。

文档导航

Documentation

Overview

Package flx provides generic stream processing with dynamic concurrency control, explicit context-aware transforms, and reusable retry, timeout, and parallel execution helpers.

Module path: github.com/ezra-sullivan/flx

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNilController                               = errors.New("flx: nil concurrency controller")
	ErrInterruptibleWorkersRequireContextTransform = errors.New("flx: WithInterruptibleWorkers/WithForcedDynamicWorkers requires MapContext/FlatMapContext")
)
View Source 
var (
	ErrInvalidRetryTimes              = errors.New("flx: retry times must be greater than 0")
	ErrNegativeRetryInterval          = errors.New("flx: retry interval must not be negative")
	ErrNegativeRetryTimeout           = errors.New("flx: retry timeout must not be negative")
	ErrNegativeAttemptTimeout         = errors.New("flx: attempt timeout must not be negative")
	ErrAttemptTimeoutRequiresRetryCtx = errors.New("flx: WithAttemptTimeout requires DoWithRetryCtx")
	ErrRetryAttemptTimeout            = errors.New("flx: retry attempt timeout")
)
View Source 
var (
	ErrCanceled        = context.Canceled
	ErrTimeout         = context.DeadlineExceeded
	ErrNilContext      = errors.New("flx: nil context")
	ErrNegativeTimeout = errors.New("flx: timeout must not be negative")
)
View Source 
var (
	ErrInvalidWindowCount    = errors.New("flx: window count must be greater than 0")
	ErrInvalidWindowDuration = errors.New("flx: window duration must be positive")
)
View Source 
var ErrInvalidErrorStrategy = errors.New("flx: invalid error strategy")
View Source 
var ErrWorkerLimitReduced = errors.New("flx: worker canceled because concurrency limit was reduced")

Functions

func DoWithRetry 

func DoWithRetry(fn func() error, opts ...RetryOption) error

func DoWithRetryCtx 

func DoWithRetryCtx(ctx context.Context, fn func(context.Context, int) error, opts ...RetryOption) error

func DoWithTimeout 

func DoWithTimeout(fn func() error, timeout time.Duration, opts ...TimeoutOption) error

func DoWithTimeoutCtx 

func DoWithTimeoutCtx(fn func(context.Context) error, timeout time.Duration, opts ...TimeoutOption) error

func Parallel 

func Parallel(fns ...func())

func ParallelErr 

func ParallelErr(fns ...func() error) error

func ParallelWithErrorStrategy 

func ParallelWithErrorStrategy(strategy ErrorStrategy, fns ...func()) error

func Reduce 

func Reduce[T, R any](s Stream[T], fn func(<-chan T) (R, error)) (R, error)

func SendContext 

func SendContext[T any](ctx context.Context, pipe chan<- T, item T) bool

Types

type ConcurrencyController 

type ConcurrencyController struct {
	// contains filtered or unexported fields
}

func NewConcurrencyController 

func NewConcurrencyController(workers int) *ConcurrencyController

func (*ConcurrencyController) ActiveWorkers 

func (c *ConcurrencyController) ActiveWorkers() int

func (*ConcurrencyController) SetWorkers 

func (c *ConcurrencyController) SetWorkers(n int)

func (*ConcurrencyController) Workers 

func (c *ConcurrencyController) Workers() int

type DynamicSemaphore 

type DynamicSemaphore struct {
	// contains filtered or unexported fields
}

func NewDynamicSemaphore 

func NewDynamicSemaphore(n int) *DynamicSemaphore

func (*DynamicSemaphore) Acquire 

func (s *DynamicSemaphore) Acquire()

func (*DynamicSemaphore) AcquireCtx 

func (s *DynamicSemaphore) AcquireCtx(ctx context.Context) error

func (*DynamicSemaphore) Cap 

func (s *DynamicSemaphore) Cap() int

func (*DynamicSemaphore) Current 

func (s *DynamicSemaphore) Current() int

func (*DynamicSemaphore) Release 

func (s *DynamicSemaphore) Release()

func (*DynamicSemaphore) Resize 

func (s *DynamicSemaphore) Resize(n int)

type ErrorStrategy 

type ErrorStrategy uint8
const (
	ErrorStrategyFailFast ErrorStrategy = iota
	ErrorStrategyCollect
	ErrorStrategyLogAndContinue
)

func (ErrorStrategy) String 

func (s ErrorStrategy) String() string

type Group added in v0.1.2 

type Group[K comparable, T any] struct {
	Key   K
	Items []T
}

type Option 

type Option func(*opOptions)

func WithDynamicWorkers 

func WithDynamicWorkers(controller *ConcurrencyController) Option

WithDynamicWorkers enables graceful dynamic resizing for the current operation. Shrinking does not interrupt workers that already hold a slot.

func WithErrorStrategy 

func WithErrorStrategy(strategy ErrorStrategy) Option

WithErrorStrategy configures how worker panic/error is handled for the current operation.

func WithForcedDynamicWorkers 

func WithForcedDynamicWorkers(controller *ConcurrencyController) Option

WithForcedDynamicWorkers enables forced dynamic resizing for the current operation. Shrinking cancels excess workers via context and only works with MapContext/FlatMapContext.

func WithInterruptibleWorkers 

func WithInterruptibleWorkers(controller *ConcurrencyController) Option

WithInterruptibleWorkers is kept as a compatibility alias for WithForcedDynamicWorkers.

func WithUnlimitedWorkers 

func WithUnlimitedWorkers() Option

WithUnlimitedWorkers spawns one worker per item for the current operation.

func WithWorkers 

func WithWorkers(workers int) Option

WithWorkers limits the current operation to a fixed number of workers.

type RetryOption 

type RetryOption func(*retryOptions)

func WithAttemptTimeout 

func WithAttemptTimeout(timeout time.Duration) RetryOption

func WithIgnoreErrors 

func WithIgnoreErrors(ignoreErrors []error) RetryOption

func WithInterval 

func WithInterval(interval time.Duration) RetryOption

func WithRetry 

func WithRetry(times int) RetryOption

func WithTimeout 

func WithTimeout(timeout time.Duration) RetryOption

type Stream 

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

func Chunk 

func Chunk[T any](s Stream[T], n int) Stream[[]T]

func Concat 

func Concat[T any](s Stream[T], others ...Stream[T]) Stream[T]

func DistinctBy 

func DistinctBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[T]

func DistinctByCount added in v0.1.2 

func DistinctByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[T]

func DistinctByWindow added in v0.1.2 

func DistinctByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[T]

func FlatMap 

func FlatMap[T, U any](s Stream[T], fn func(T, chan<- U), opts ...Option) Stream[U]

func FlatMapContext 

func FlatMapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U), opts ...Option) Stream[U]

func FlatMapContextErr 

func FlatMapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U) error, opts ...Option) Stream[U]

func FlatMapErr 

func FlatMapErr[T, U any](s Stream[T], fn func(T, chan<- U) error, opts ...Option) Stream[U]

func From 

func From[T any](generate func(chan<- T)) Stream[T]

func FromChan 

func FromChan[T any](source <-chan T) Stream[T]

func GroupBy 

func GroupBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[[]T]

func GroupByCount added in v0.1.2 

func GroupByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[Group[K, T]]

func GroupByWindow added in v0.1.2 

func GroupByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[Group[K, T]]

func Map 

func Map[T, U any](s Stream[T], fn func(T) U, opts ...Option) Stream[U]

func MapContext 

func MapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) U, opts ...Option) Stream[U]

func MapContextErr 

func MapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) (U, error), opts ...Option) Stream[U]

func MapErr 

func MapErr[T, U any](s Stream[T], fn func(T) (U, error), opts ...Option) Stream[U]

func Values 

func Values[T any](items ...T) Stream[T]

func (Stream[T]) AllMatch 

func (s Stream[T]) AllMatch(predicate func(T) bool) bool

func (Stream[T]) AllMatchErr 

func (s Stream[T]) AllMatchErr(predicate func(T) bool) (bool, error)

func (Stream[T]) AnyMatch 

func (s Stream[T]) AnyMatch(predicate func(T) bool) bool

func (Stream[T]) AnyMatchErr 

func (s Stream[T]) AnyMatchErr(predicate func(T) bool) (bool, error)

func (Stream[T]) Buffer 

func (s Stream[T]) Buffer(n int) Stream[T]

func (Stream[T]) Collect 

func (s Stream[T]) Collect() []T

func (Stream[T]) CollectErr 

func (s Stream[T]) CollectErr() ([]T, error)

func (Stream[T]) Concat 

func (s Stream[T]) Concat(others ...Stream[T]) Stream[T]

func (Stream[T]) Count 

func (s Stream[T]) Count() (count int)

func (Stream[T]) CountErr 

func (s Stream[T]) CountErr() (int, error)

func (Stream[T]) Done 

func (s Stream[T]) Done()

func (Stream[T]) DoneErr 

func (s Stream[T]) DoneErr() error

func (Stream[T]) Err 

func (s Stream[T]) Err() error

func (Stream[T]) Filter 

func (s Stream[T]) Filter(fn func(T) bool, opts ...Option) Stream[T]

func (Stream[T]) First 

func (s Stream[T]) First() (T, bool)

func (Stream[T]) FirstErr 

func (s Stream[T]) FirstErr() (T, bool, error)

func (Stream[T]) ForAll 

func (s Stream[T]) ForAll(fn func(<-chan T))

func (Stream[T]) ForAllErr 

func (s Stream[T]) ForAllErr(fn func(<-chan T)) error

func (Stream[T]) ForEach 

func (s Stream[T]) ForEach(fn func(T))

func (Stream[T]) ForEachErr 

func (s Stream[T]) ForEachErr(fn func(T)) error

func (Stream[T]) Head 

func (s Stream[T]) Head(n int64) Stream[T]

func (Stream[T]) Last 

func (s Stream[T]) Last() (T, bool)

func (Stream[T]) LastErr 

func (s Stream[T]) LastErr() (T, bool, error)

func (Stream[T]) Max 

func (s Stream[T]) Max(less func(T, T) bool) (T, bool)

func (Stream[T]) MaxErr 

func (s Stream[T]) MaxErr(less func(T, T) bool) (T, bool, error)

func (Stream[T]) Min 

func (s Stream[T]) Min(less func(T, T) bool) (T, bool)

func (Stream[T]) MinErr 

func (s Stream[T]) MinErr(less func(T, T) bool) (T, bool, error)

func (Stream[T]) NoneMatch 

func (s Stream[T]) NoneMatch(predicate func(T) bool) bool

func (Stream[T]) NoneMatchErr 

func (s Stream[T]) NoneMatchErr(predicate func(T) bool) (bool, error)

func (Stream[T]) Parallel 

func (s Stream[T]) Parallel(fn func(T), opts ...Option)

func (Stream[T]) ParallelErr 

func (s Stream[T]) ParallelErr(fn func(T) error, opts ...Option) error

func (Stream[T]) Reverse 

func (s Stream[T]) Reverse() Stream[T]

func (Stream[T]) Skip 

func (s Stream[T]) Skip(n int64) Stream[T]

func (Stream[T]) Sort 

func (s Stream[T]) Sort(less func(T, T) bool) Stream[T]

func (Stream[T]) Tail 

func (s Stream[T]) Tail(n int64) Stream[T]

type TimeoutOption 

type TimeoutOption func() context.Context

func WithContext 

func WithContext(ctx context.Context) TimeoutOption

type WorkerError 

type WorkerError struct {
	Err error
}

func (*WorkerError) Error 

func (e *WorkerError) Error() string

func (*WorkerError) Unwrap 

func (e *WorkerError) Unwrap() error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.