pipeline

package

v0.3.7 Latest Latest Go to latest Published: Aug 8, 2026 License: Apache-2.0 Imports: 2 Imported by: 0

Details

Package pipeline 提供类型安全的多阶段流水线:数据经过多个 Stage 顺序处理, 每个 Stage 可配置并发 worker 数(扇出)、带背压(有界 channel)。

典型场景:

与 stream/router 区别:

纯标准库、泛型。

This section is empty.

This section is empty.

func Merge[T any](ctx context.Context, channels ...<-chan T) <-chan T

Merge 扇入:合并多个 channel 到一个。所有输入关闭后输出关闭。

func Pipe[In, Out any](ctx context.Context, input <-chan In, stage Stage[In, Out]) (<-chan Out, <-chan error)

Pipe 连接两个阶段:从 input channel 读取,经 stage 处理,输出到新 channel。返回输出 channel 和 error channel。当 input 关闭且所有 worker 完成后,输出 channel 关闭。

func Run[In, Out any](ctx context.Context, source <-chan In, stage Stage[In, Out]) ([]Out, error)

Run 执行完整的单阶段 pipeline:从 source 读,经 stage 处理,收集结果。阻塞直到 source 关闭或 ctx 取消或出错。

func Source[T any](ctx context.Context, items []T) <-chan T

Source 将切片转为 channel(方便测试和简单场景)。

func Split[T any](ctx context.Context, input <-chan T, n int) []<-chan T

Split 扇出:将一个 channel 复制到 n 个输出 channel。每条数据只发到一个输出(轮询分发)。

type ProcessFunc[In, Out any] func(ctx context.Context, in In, emit func(Out)) error

ProcessFunc 阶段处理函数。输入一条数据,输出零或多条(通过 emit)。返回 error 表示致命错误,将终止整个 pipeline。

type Stage[In, Out any] struct {
	Process ProcessFunc[In, Out]
	Workers int // 并发 worker 数,默认 1
	BufSize int // 输出 channel 缓冲大小,默认 0(同步)
}

Stage 表示流水线的一个阶段配置。