flx

package module
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Mar 22, 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
  • 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

动态并发

优雅缩容

缩容时不打断已运行 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。

错误处理建议

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

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

业务代码里推荐优先使用 *Err 变体，这样错误边界最清楚：

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

与 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 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 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 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 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.