stream

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: MIT Imports: 5 Imported by: 0

README

go-stream

CI codecov Quality Gate Status govulncheck OpenSSF Scorecard CodeQL Go Reference

English | 简体中文

A Go 1.27 generics implementation of the Java Stream API — built on the newly introduced generic methods feature, enabling natural, fluent stream processing in native Go for the first time:

stream.Of(1, 2, 3, 4, 5).
    Filter(func(n int) bool { return n%2 == 1 }).
    Map(func(n int) int { return n * n }).
    ToSlice() // [1 9 25]

Before Go 1.27, methods could not declare their own type parameters, so chained APIs like Map[U] could only be written as package-level functions (stream.Map(s, f)) with no fluent chaining. With generic methods, methods on Stream[T] can carry their own type parameters, enabling fully chained declarations with static type migration along the pipeline.

Features

  • Lazy pipelines: intermediate operations only declare the pipeline without triggering traversal; a terminal operation triggers a single fused evaluation pass
  • Generic methods: method-level type parameters such as Map[U]/Zip[U, R]/Collect[A, R] let element types migrate statically along the pipeline
  • Single-pass fusion: stateless operators fuse into a single pass (Sink chain) at evaluation time; stateful operators materialize in segments
  • Short-circuit evaluation: Limit/First/AnyMatch/TakeWhile and friends stop source traversal as soon as the condition is met (safe for infinite streams)
  • Errors as values: expected errors (IO source failures, MapErr family callback errors) propagate as error values — first error short-circuits, partial results are preserved, query via Err(); unrecoverable misuses (double consumption, nil callbacks) panic
  • Composition over inheritance: Java's abstract class hierarchy (AbstractPipeline/StatelessOp/StatefulOp) is translated into "struct embedding + constructors + injected function values" with no simulated inheritance
  • Zero third-party dependencies: no third-party runtime dependencies in v1

Installation

go get github.com/JayceChant/go-stream

Requires go 1.27+ (relies on the generic methods feature).

Quick Start

import (
    "github.com/JayceChant/go-stream"
    "github.com/JayceChant/go-stream/collector" // collector subpackage (as needed)
)

// 1. Build from containers (lazy, no traversal yet)
s := stream.FromSlice(data)          // zero-copy reference
r := stream.Range(0, 100)            // integer range [0, 100)
g := stream.Generate(func() int { return 42 }) // infinite generator

// 2. Intermediate operations (return a new Stream, chainable)
s.Filter(p).Map(f).Sorted(cmp).Limit(10)

// 3. Terminal operations (trigger a single evaluation, consume the stream)
s.ToSlice()
s.Count()
s.AnyMatch(p)
s.Collect(collector.GroupingBy(keyOf, valOf))

More runnable examples: example_test.go (verified by go test) and the example/ directory — six standalone, copy-paste-ready programs covering the full API surface:

go -C example run ./basics      # sources → intermediate → terminal operations
go -C example run ./collectors  # collector family + custom Collector (TopN)
go -C example run ./numeric     # numeric aggregation, Scan, infinite sources, Zip/Chunk/Enumerate
go -C example run ./errors      # errors-as-value model (FromFunc/MapErr family/Err())
go -C example run ./parallel    # Parallel(n)/Unordered, order-preserving merge, auto fallback
go -C example run ./lifecycle   # OnClose/Close resource management, Cache replayable factory

example/ is a separate Go module (not part of the library's tests or coverage) so each file can be copied into your project as-is.

API Overview

Category APIs
Construction Of FromSlice FromSeq FromChannel FromMap FromFunc Generate Iterate Range Concat Empty
Stateless intermediate Filter Map FlatMap FlatMapSeq Peek TakeWhile DropWhile
Err variants MapErr FilterErr FlatMapErr PeekErr
Stateful intermediate Limit Skip Sorted DistinctBy Reverse Scan
Parallelism control Parallel(n) Sequential() Unordered()
Package-level intermediate Distinct Sorted (natural order) Chunk Enumerate
Two-stream Zip
Lifecycle OnClose(f) Close() Cache(s) (replayable factory)
Terminal ForEach ForEachUntil ToSlice Count Reduce ReduceOpt Collect First FindAny AnyMatch AllMatch NoneMatch Min Max Err
Collectors (subpackage collector) ToSlice ToSet ToMap ToMapMerge GroupingBy Joining Counting Reducing Mapping
Collectors (root package) Summing (relies on the Number constraint)
Package-level aggregation Sum Avg Contains Min Max

For the full reference and examples, see docs/api.md.

Comparison with Java Stream

Java go-stream Notes
Stream<T> (interface) *Stream[T] (concrete struct) In Go 1.27 interface methods cannot declare type parameters; generic methods must live on concrete types
stream.of(...) / Arrays.stream stream.Of(...) / stream.FromSlice
Collectors.toList() collector.ToSlice[T]()
Collectors.toMap collector.ToMap / ToMapMerge Key conflicts: last-wins (aligned with Go map conventions); use ToMapMerge for custom merging
Collectors.groupingBy collector.GroupingBy Preserves encounter order within groups
Comparator func(a, b T) int Aligned with the standard library's slices.SortFunc/cmp.Compare conventions
IntStream specializations Generics + Number/cmp.Ordered constraints Go generics have zero boxing; no specialization needed
stream.parallel() Parallel(n) / Sequential() TrySplit splitting + goroutines; automatically falls back to sequential after short-circuit terminals or materializing operators
stream.unordered() Unordered() Clears the SpOrdered flag; under parallelism, shard results are pushed as they complete (streaming merge)
stream.onClose(f) / close() OnClose(f) / Close() Triggered automatically at the end of evaluation (including short-circuit/error/panic paths); explicit close is idempotent; callback errors are queryable via Err()
Exception propagation Errors as values (Err()/MapErr family) Aligned with Go's official error style
stream.distinct() DistinctBy[K comparable](key) method / Distinct package-level A method's own type parameters may carry the comparable constraint (keys are compile-time comparable, zero boxing); Distinct constrains the element T itself, and methods cannot constrain the receiver's T, so it stays package-level

Design Highlights

  • Sink push chain: at evaluation time, sinks are wrapped in reverse starting from the terminal operation (Accept(t) bool merges Java's cancellationRequested); the data source pushes elements through the entire chain in a single pass
  • Segmented evaluation: stateful operators such as Sorted/Skip first drive upstream to materialize []T, then transform and replay; Limit supports short-circuit collection from infinite sources; Skip(0) returns the original stream as a true no-op (no materialization, flags passthrough)
  • Flag propagation: SpSized/SpOrdered/SpSorted/SpDistinct propagate along the pipeline (e.g. Map preserves Sized 1:1 so downstream can preallocate), informing parallel splitting decisions
  • Error model: modeled after the bufio.Scanner.Err() convention — on error, terminal operations return the accumulated partial results, and Err() returns the first error

See docs/design.md for architecture details.

Performance

Filter+Map+ToSlice vs. a hand-written for loop (a realistic scenario including strconv.Itoa):

Scale Pipeline Hand-written for Overhead
1e2 ~2.8 μs ~1.0 μs 2.8x
1e4 ~0.48 ms ~0.18 ms 2.6x
1e6 ~35 ms ~22 ms 1.6x

Target of <3x met (AMD Ryzen 5 7535U, benchtime 300ms; reproduce with go test -bench . -run '^$').

Roadmap

  • v1 sequential evaluation engine, full operator set, Collector system, errors-as-values model
  • Parallel evaluation Parallel(n) / Sequential(): recursive TrySplit splitting + goroutine-parallel execution + Collector.Combiner merging; order-preserving merge by shard order (Ordered); automatic fallback to sequential after short-circuit terminals or materializing operators (correctness first); measured speedup of ~3.3x (4 shards) on CPU-bound workloads
  • v1.x: onClose/resource management (OnClose(f) triggered automatically at end of evaluation + idempotent explicit Close()), replayable streams (Cache(s) factory: materialize once, produce a brand-new one-shot stream each time without breaking the one-shot model), Unordered streaming merge (Unordered() clears the order flag; under parallelism shards push results as they complete, reducing end-to-end latency)

License

MIT

Documentation

Overview

Package stream 提供基于 Go 1.27 泛型方法的 Java Stream 风格流式处理库。

流（Stream）不是数据结构，而是从数据源（slice、channel、迭代器、生成器等） 到结果之间的惰性管道：中间操作仅追加管道阶段不触发遍历，终止操作触发一次 单遍求值并返回新容器或聚合值。典型用法： 

stream.Of(1, 2, 3).Filter(func(v int) bool { return v > 1 }).Map(strconv.Itoa).ToSlice()

详细设计（架构、错误模型、API 详案）见项目 spec/spec.md。

Example (ErrorAsValue)

错误即值：FromFunc 源失败时部分结果保留，Err() 查询首错。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	n := 0
	s := stream.FromFunc(func() (int, bool, error) {
		n++
		if n >= 4 {
			return 0, false, fmt.Errorf("第 %d 次读取失败", n)
		}
		return n, true, nil
	})
	got := s.ToSlice()
	fmt.Println(got)
	fmt.Println(s.Err())
}

Output:
[1 2 3]
第 4 次读取失败
Example (FromMap)

map 源：FromMap 产出 KV 键值对。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	m := map[string]int{"one": 1, "two": 2}
	total := 0
	stream.FromMap(m).ForEach(func(kv stream.KV[string, int]) {
		total += kv.Value
	})
	fmt.Println(total)
}

Output:
3
Example (InfiniteWithLimit)

无限流 + 短路：Generate 搭配 Limit。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	i := 0
	got := stream.Generate(func() int { i++; return i * i }).
		Limit(5).
		ToSlice()
	fmt.Println(got)
}

Output:
[1 4 9 16 25]
Example (MapTypeChange)

类型迁移：Map 把 int 流变为 string 流（编译期静态类型检查）。 

package main

import (
	"fmt"
	"strconv"

	"github.com/JayceChant/go-stream"
	"github.com/JayceChant/go-stream/collector"
)

func main() {
	got := stream.Of(1, 2, 3).
		Map(strconv.Itoa).
		Collect(collector.Joining(func(s string) string { return s }, "-"))
	fmt.Println(got)
}

Output:
1-2-3
Example (Numeric)

数值聚合：包级 Sum/Avg（方法无法追加 Number 约束）。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	fmt.Println(stream.Sum(stream.Range(1, 101)))
	fmt.Println(stream.Avg(stream.Range(1, 4)))
}

Output:
5050
2
Example (SortedDistinctPage)

排序去重分页组合。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	got := stream.Of(5, 3, 1, 3, 5, 2, 1).
		Sorted(func(a, b int) int { return a - b }).
		DistinctBy(func(n int) int { return n }).
		Skip(1).
		Limit(3).
		ToSlice()
	fmt.Println(got)
}

Output:
[2 3 5]
Example (Zip)

双流拉链：Zip 取短。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	got := stream.Of("a", "b", "c").
		Zip(stream.Range(1, 100), func(s string, i int) string {
			return fmt.Sprintf("%s%d", s, i)
		}).
		ToSlice()
	fmt.Println(got)
}

Output:
[a1 b2 c3]

Index

Examples

Constants

This section is empty.

Variables

This section is empty.

Functions

func Avg 

func Avg[N Number](s *Stream[N]) N

Avg 数值平均（空流返回 0）。

func Cache 

func Cache[T any](s *Stream[T]) func() *Stream[T]

Cache 把一次性流 s 转为可重放工厂：首次调用工厂时求值 s 一次并物化 全部元素，此后每次调用返回全新的独立流（FromSlice 共享底层数组， 零拷贝）。

一次性模型保持：s 在首次调用时被消费（工厂从未被调用则 s 仍可用）； 工厂产物每次也是一次性流。

错误语义：物化期 s 出错 → 首错记忆进工厂，此后每次调用返回携带该 错误的空流（任何终止操作得空结果，Err() 返回该错误）。

func Contains 

func Contains[T comparable](s *Stream[T], target T) bool

Contains 判断流中是否含有目标元素（短路）。

func Max 

func Max[T cmp.Ordered](s *Stream[T]) (T, bool)

Max 依自然序取最大（空流返回零值与 false）。

func Min 

func Min[T cmp.Ordered](s *Stream[T]) (T, bool)

Min 依自然序取最小（空流返回零值与 false）。

func Sum 

func Sum[N Number](s *Stream[N]) N

Sum 数值求和。

func Summing 

func Summing[N Number]() collector.Collector[N, *N, N]

Summing 数值求和收集器（Number 约束的便捷形态）。 依赖根包 Number 约束故留根包；与子包 collector.Collector 兼容。

Types

type Characteristics 

type Characteristics uint16

Characteristics 是数据源结构特征的位标志。

求值器可依据特征位省略不必要的工作（如源已 Sorted 则跳过排序）； 后续并行实现（见 spec 后续 TODO）将依赖 Sized/SubSized 做均衡拆分。 

const (
	// SpSized 表示源可精确报告元素数量（EstimateSize 返回精确值）。
	SpSized Characteristics = 1 << iota
	// SpOrdered 表示源具有确定的相遇顺序，遍历与拆分必须保持该顺序。
	SpOrdered
	// SpSubSized 表示 TrySplit 产生的子源同样可精确报告大小（仅在 SpSized 时有意义）。
	SpSubSized
	// SpSorted 表示源元素按某比较器有序。
	SpSorted
	// SpDistinct 表示源元素两两不重复。
	SpDistinct
)

type Float 

type Float interface {
	~float32 | ~float64
}

Float 约束全部浮点类型。

type Integer 

type Integer interface {
	~int | ~int8 | ~int16 | ~int32 | ~int64 |
		~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64 | ~uintptr
}

Integer 约束全部有符号与无符号整数类型，供 Range 等构造使用。

type KV 

type KV[K comparable, V any] struct {
	Key   K
	Value V
}

KV 是键值对元素，供 FromMap（map 源）、Enumerate（索引配对）等场景使用。

type Number 

type Number interface {
	Integer | Float
}

Number 约束全部数值类型（整数与浮点），供 Sum/Avg 等聚合使用。

type Sink 

type Sink[T any] interface {
	// Begin 在元素推送开始前调用一次，size 为源的估计元素数（未知为 -1），
	// 便于实现预分配容量等优化。
	Begin(size int64)
	// Accept 接收一个元素；返回 false 表示请求取消（短路），
	// 引擎将立即停止推动源并调用 End。
	Accept(t T) bool
	// End 在元素推送结束后调用一次（无论正常耗尽还是短路取消）。
	// 实现应在此完成收尾（如排序后统一输出）。
	End()
}

Sink 是推送式消费者接口，是求值期间元素流动的通道。

求值引擎从终止操作出发，将各级中间操作的 Sink 逐级反向包装成一条链， 再由数据源单遍推动元素流过整条链（单遍融合）。

注意：接口方法不得声明自身类型参数（Go 1.27 语言限制），因此本接口 只使用 Stream 的元素类型 T。

type Splitterator 

type Splitterator[T any] interface {
	// TryAdvance 尝试推进到下一个元素：存在则以该元素调用 f 一次并返回 true；
	// f 返回 false 表示消费方请求短路（实现不得再调用 f）；元素耗尽返回 false。
	TryAdvance(f func(T) bool) bool
	// ForEachRemaining 遍历剩余全部元素并逐个调用 f；f 返回 false 时提前结束。
	// 默认可基于 TryAdvance 实现，高效源应自行提供。
	ForEachRemaining(f func(T) bool)
	// TrySplit 尝试把剩余元素分裂出一个新的 Splitterator 供并行处理；
	// 不可分裂（或分裂无益）时返回 nil。串行实现恒为 nil。
	TrySplit() Splitterator[T]
	// EstimateSize 返回剩余元素估计数；未知返回 -1。Sized 特征下为精确值。
	EstimateSize() int64
	// Characteristics 返回本源的特征位集合。
	Characteristics() Characteristics
}

Splitterator 是流的数据源抽象：可逐个推进、可整体遍历、可分裂（为并行预留）。

概念对应 Java 的 Spliterator（Go 侧按拼写惯例写作 Splitterator）。 一个 Splitterator 只服务于一次求值：遍历与分裂都会消耗其元素。

type Stream 

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

Stream 是对外公开的流类型：一条惰性管道的句柄。

通过包级构造函数（Of/FromSlice/FromSeq/Range 等）创建；中间操作以泛型方法 追加阶段并返回新的 *Stream（如 Map[U]）；终止操作触发一次求值并消费本流。 同一实例仅可被链接或消费一次，重复使用将 panic（编程错误，不可恢复）。

func Chunk 

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

Chunk 把连续元素切分为定长分组（尾组可能不足 n）。n <= 0 panic。

包级函数形态：Go 1.27 泛型方法返回 Stream[[]T]（T 的派生类型）会触发 实例化循环（T → []T → [][]T → ...），只能以包级函数提供。 有状态单遍（跨元素缓冲）→ 并行降级（splitN=nil）。

func Concat 

func Concat[T any](a, b *Stream[T]) *Stream[T]

Concat 串联两条流：先耗尽 a 再消费 b（a、b 均被标记消费）。 两段推入同一下游 sink，共用一次 Begin/End（经 suppressEnd 吞掉 a 段 下传的 End，由 b 段统一收尾；b 段自带的 Begin 被 skipBegin 吞掉）。

func Distinct 

func Distinct[T comparable](s *Stream[T]) *Stream[T]

Distinct 依据元素自身可比较性去重（保留首见，保持遇序）。 包级函数形态：Go 方法无法对 T 追加 comparable 约束。

func Empty 

func Empty[T any]() *Stream[T]

Empty 构建空流。

func Enumerate 

func Enumerate[T any](s *Stream[T]) *Stream[KV[int, T]]

Enumerate 为元素附加从 0 开始的索引，产出 KV[int, T] （对应 Go for i, v := range 习惯）。

包级函数形态：同 Chunk，泛型方法返回 Stream[KV[int, T]] 触发实例化循环。 有状态单遍（递增索引）→ 并行降级（splitN=nil）。

func FromChannel 

func FromChannel[T any](ch <-chan T) *Stream[T]

FromChannel 基于 channel 构建流（阻塞拉取直到通道关闭）。

func FromFunc 

func FromFunc[T any](next func() (T, bool, error)) *Stream[T]

FromFunc 基于拉式函数构建流：next 返回 (元素, 是否还有, 错误)。 适配 IO/解析等可失败源；首错经 Err() 查询，出错时保留已产出的部分结果。

func FromMap 

func FromMap[K comparable, V any](m map[K]V) *Stream[KV[K, V]]

FromMap 基于 map 构建流，产出 KV 键值对元素。 map 遍历顺序不确定，故本源为 Unordered（不声明 SpOrdered—— Task 10 修正：此前经 newSeqSp 误置 SpOrdered，与本源语义矛盾）。

func FromSeq 

func FromSeq[T any](seq iter.Seq[T]) *Stream[T]

FromSeq 基于 Go 1.23 push 迭代器构建流（一次性：seq 无法暂停复用）。

func FromSlice 

func FromSlice[T any](s []T) *Stream[T]

FromSlice 基于 slice 构建流（零拷贝，直接引用原切片）。

func Generate 

func Generate[T any](f func() T) *Stream[T]

Generate 以无限生成函数构建流（须配合 Limit 等短路算子终止）。

func Iterate 

func Iterate[T any](seed T, next func(T) T) *Stream[T]

Iterate 以种子与后继函数构建无限流：seed, f(seed), f(f(seed)), ...。

func Of 

func Of[T any](xs ...T) *Stream[T]

Of 以可变参数构建流。

Example

基础链式：构造 → 中间操作 → 终止操作。 

package main

import (
	"fmt"

	"github.com/JayceChant/go-stream"
)

func main() {
	got := stream.Of(1, 2, 3, 4, 5).
		Filter(func(n int) bool { return n%2 == 1 }).
		Map(func(n int) int { return n * n }).
		ToSlice()
	fmt.Println(got)
}

Output:
[1 9 25]

func Range 

func Range[I Integer](start, stop I) *Stream[I]

Range 构建整数区间流 [start, stop)（左闭右开，步长 1）。

func Sorted 

func Sorted[T cmp.Ordered](s *Stream[T]) *Stream[T]

Sorted 依自然序（cmp.Ordered）稳定排序。 包级函数形态：方法无法对 T 追加 cmp.Ordered 约束。

func (*Stream[T]) AllMatch 

func (s *Stream[T]) AllMatch(p func(T) bool) bool

AllMatch 是否全部元素满足 p（短路：遇首个不满足返回 false；空流 true）。

func (*Stream[T]) AnyMatch 

func (s *Stream[T]) AnyMatch(p func(T) bool) bool

AnyMatch 是否存在满足 p 的元素（短路：命中即返回 true）。

func (*Stream[T]) Close 

func (s *Stream[T]) Close() error

Close 显式关闭本流：立即触发回调链（幂等——重复调用、求值后再 Close 均不重复执行；未求值的流也可关闭）。返回回调链首错，并记入错误槽 （Err() 可查询）。

func (*Stream[T]) Collect 

func (s *Stream[T]) Collect[A, R any](c collector.Collector[T, A, R]) R

Collect 以自定义收集器汇聚元素（泛型方法，支持 A→R 类型迁移）。 并行流：片级独立累积，按分片序以 Combiner 合并，Finisher 收尾。 收集器族见子包 collector（stream/collector）。

func (*Stream[T]) Count 

func (s *Stream[T]) Count() int64

Count 返回元素总数（并行流片内计数、片序求和）。

func (*Stream[T]) DistinctBy 

func (s *Stream[T]) DistinctBy[K comparable](key func(T) K) *Stream[T]

DistinctBy 依据 key 函数去重：每组同 key 仅保留首个遇到的元素（保遇序）。 K 须满足 comparable：键为具体不可比较类型（slice/map/func 等）时编译期即报错。 逃生口：K 显式取 any（接口满足 comparable）仍可编译，动态类型不可比较时 在求值时 panic（用户契约，同 map 键语义）。

func (*Stream[T]) DropWhile 

func (s *Stream[T]) DropWhile(p func(T) bool) *Stream[T]

DropWhile 丢弃首批满足 p 的元素，之后全部放行。 有状态单遍（done 门闸）→ 并行降级（splitN=nil）。

func (*Stream[T]) Err 

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

Err 返回最近一次由本流发起的终止求值的首错（错误即值模型）。 无错误返回 nil；未求值前调用亦返回 nil。

func (*Stream[T]) Filter 

func (s *Stream[T]) Filter(p func(T) bool) *Stream[T]

Filter 保留满足谓词 p 的元素。

func (*Stream[T]) FilterErr 

func (s *Stream[T]) FilterErr(p func(T) (bool, error)) *Stream[T]

FilterErr 带错误返回的 Filter。

func (*Stream[T]) FindAny 

func (s *Stream[T]) FindAny(p func(T) bool) (T, bool)

FindAny 寻找任一满足 p 的元素（短路）。顺序流下等价于 First + Filter。

func (*Stream[T]) First 

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

First 返回首个元素（短路：取到即停）；空流返回 (零值, false)。

func (*Stream[T]) FlatMap 

func (s *Stream[T]) FlatMap[U any](f func(T) []U) *Stream[U]

FlatMap 将每个元素经 f 展开为子序列并依次输出。

func (*Stream[T]) FlatMapErr 

func (s *Stream[T]) FlatMapErr[U any](f func(T) ([]U, error)) *Stream[U]

FlatMapErr 带错误返回的 FlatMap。

func (*Stream[T]) FlatMapSeq 

func (s *Stream[T]) FlatMapSeq[U any](f func(T) iter.Seq[U]) *Stream[U]

FlatMapSeq 与 FlatMap 相同，但展开函数返回 iter.Seq（支持惰性子序列）。

func (*Stream[T]) ForEach 

func (s *Stream[T]) ForEach(f func(T))

ForEach 对每个元素执行 f（并行流下按相遇序合并、f 在发起 goroutine 串行调用；Unordered 流例外：按分片完成序推入，顺序不保证）。

func (*Stream[T]) ForEachUntil 

func (s *Stream[T]) ForEachUntil(f func(T) bool)

ForEachUntil 对每个元素执行 f；f 返回 false 时提前终止。

func (*Stream[T]) Limit 

func (s *Stream[T]) Limit(n int64) *Stream[T]

Limit 截取前 n 个元素（n == 0 得空流；无限源可借此终止；n < 0 panic）。

func (*Stream[T]) Map 

func (s *Stream[T]) Map[U any](f func(T) U) *Stream[U]

Map 将每个元素经 f 变换为新类型 U（泛型方法，元素类型迁移）。 1:1 变换：保留 SpSized（下游可按 size 预分配），仅清 SpSorted/SpDistinct。

func (*Stream[T]) MapErr 

func (s *Stream[T]) MapErr[U any](f func(T) (U, error)) *Stream[U]

MapErr 带错误返回的 Map：f 出错时记录首错并终止求值。 1:1 变换：保留 SpSized，仅清 SpSorted/SpDistinct。

func (*Stream[T]) Max 

func (s *Stream[T]) Max(cmp func(a, b T) int) (T, bool)

Max 返回最大元素（依 cmp）；空流返回 (零值, false)。

func (*Stream[T]) Min 

func (s *Stream[T]) Min(cmp func(a, b T) int) (T, bool)

Min 返回最小元素（依 cmp）；空流返回 (零值, false)。

func (*Stream[T]) NoneMatch 

func (s *Stream[T]) NoneMatch(p func(T) bool) bool

NoneMatch 是否无元素满足 p（空流 true）。

func (*Stream[T]) OnClose 

func (s *Stream[T]) OnClose(f func() error) *Stream[T]

OnClose 注册资源清理回调 f：返回携带回调链的新流（中间操作语义： 消费本流，已注册的回调链一并继承）。f 为 nil 时 panic（编程错误）。

触发时机：新流（或其任一下游）的终止求值结束时自动触发一次—— 正常耗尽、短路与错误值路径均触发，用户回调 panic 的展开路径亦触发； 求值前显式调用过 Close 则以显式关闭为准。多个回调按注册顺序执行； 任一出错记录首错（不 panic，可经 Err() 查询）。

幂等保证：每个物理回调以 sync.Once 包装——无论经由求值自动触发、 任一 stage 实例的显式 Close、还是组合流（Concat/Zip 继承合并后的 回调链）触发，均恰好执行一次。

func (*Stream[T]) Parallel 

func (s *Stream[T]) Parallel(n int) *Stream[T]

Parallel 声明后续求值以最多 n 个分片并行（中间操作语义：消费上游， 返回携带并行标志的新流）。n <= 1 或不可分源/含降级算子的管道自动串行。

func (*Stream[T]) Peek 

func (s *Stream[T]) Peek(f func(T)) *Stream[T]

Peek 对每个元素施加副作用 f（不改变元素，常用于调试观察）。 并行流下 f 在分片 goroutine 内执行，观察顺序不保证（需保序请用 ForEach）。

func (*Stream[T]) PeekErr 

func (s *Stream[T]) PeekErr(f func(T) error) *Stream[T]

PeekErr 带错误返回的 Peek。

func (*Stream[T]) Reduce 

func (s *Stream[T]) Reduce(identity T, op func(T, T) T) T

Reduce 以 identity 为初值折叠全部元素（并行流片内折叠、片序合并）。

func (*Stream[T]) ReduceOpt 

func (s *Stream[T]) ReduceOpt(op func(T, T) T) (T, bool)

ReduceOpt 无初值折叠：空流返回 (零值, false)。

func (*Stream[T]) Reverse 

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

Reverse 反转元素顺序。

func (*Stream[T]) Scan 

func (s *Stream[T]) Scan[U any](seed U, f func(U, T) U) *Stream[U]

Scan 滚动累积（前缀和式）：输出 seed, f(seed,x1), f(f(seed,x1),x2), ... 输出个数与输入相同（含初值，比 Java 无对应物的常见 Go 实现多一项）。 有状态单遍（滚动 acc）→ 并行降级（splitN=nil）。

func (*Stream[T]) Sequential 

func (s *Stream[T]) Sequential() *Stream[T]

Sequential 还原串行求值（抵消上游 Parallel 声明）。

func (*Stream[T]) Skip 

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

Skip 跳过前 n 个元素，输出其余（n == 0 恒等返回原流，不物化、特征位透传； n < 0 panic）。恒等返回时不标记上游 consumed，原流仍可继续链接。

func (*Stream[T]) Sorted 

func (s *Stream[T]) Sorted(cmp func(a, b T) int) *Stream[T]

Sorted 按比较器 cmp 升序稳定排序（cmp 负/零/正 表示小于/等于/大于）。

func (*Stream[T]) TakeWhile 

func (s *Stream[T]) TakeWhile(p func(T) bool) *Stream[T]

TakeWhile 保留首批满足 p 的元素，遇到首个不满足即终止（短路）。

func (*Stream[T]) ToSlice 

func (s *Stream[T]) ToSlice() []T

ToSlice 收集全部元素为新切片（并行流按相遇序合并进同一终端）。

func (*Stream[T]) Unordered 

func (s *Stream[T]) Unordered() *Stream[T]

Unordered 声明后续求值不依赖相遇顺序（清除 SpOrdered 特征位，纯标志 stage，不改变元素流与并行度）。对应 Java BaseStream.unordered()。

语义效果：并行求值下分片结果按完成序流式并入终端（先完成先推）， 降低端到端延迟；结果集合与串行一致，顺序不保证（本就是无序语义）。 仅 ToSlice/ForEach/Min/Max（元素级）与 Collect（Combiner 按完成序合并） 参与流式合并；Count/Reduce 仍按片序聚合（结果不受影响）。

func (*Stream[T]) Zip 

func (s *Stream[T]) Zip[U, R any](other *Stream[U], f func(T, U) R) *Stream[R]

Zip 把两条流按位置配对：以 f 合并本流元素与 other 对应元素， 任一流耗尽即终止（取短）。两条流均被标记消费。

Source Files

View all Source files

Directories

Path Synopsis
Package collector 提供流式汇聚的收集器族（Collector 及其预置实现）。
 Package collector 提供流式汇聚的收集器族（Collector 及其预置实现）。

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.