stream

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: MIT Imports: 6 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
  • Java 25 parity highlights: outbound iterator adaptation ToSeq() iter.Seq[T] (range-over-func interop), collector composition ecosystem (GroupingByDownstream/PartitioningBy/Teeing/Filtering/FlatMapping/CollectingAndThen/MinBy/MaxBy), sliding window WindowSliding, single-pass statistics Summary/Summarizing (SummaryStats), and convenience sources RangeClosed/OfNonZero
  • 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) → *NumberStream (numeric narrowing, see NumberStream below)
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 — seven 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
go -C example run ./extensions  # Java 25 parity: ToSeq/collector composition/WindowSliding/Summary/RangeClosed/OfNonZero

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.

Implementation Comparison

Style

The same task — keep the positive amounts, sort them in descending order, take the top 3, and format them as price strings. Sorted and Limit are stateful operators that force a materialization point mid-pipeline, and the order is significant: sorting must precede Limit (the top 3 of a sorted sequence), formatting must follow it. Hand-rolled code has no choice but to split into two loops:

Plain Go, no library:

// Plain Go: fine for a one-off loop, but the stateful steps dissolve the
// pipeline into two loops plus an in-place sort.
var amounts []int
for _, n := range orders { // loop 1: only the stateless Filter fits here
    if n > 0 {
        amounts = append(amounts, n)
    }
}
slices.SortFunc(amounts, func(a, b int) int { return b - a }) // materialization point: needs every element (unstable, same contract as Sorted)
var top []string
for i, n := range amounts { // loop 2: top 3 and formatting can only wait here
    if i >= 3 {
        break
    }
    top = append(top, fmt.Sprintf("$%d", n))
}

Functional stream, pre-Go 1.27 (no generic methods):

// Pre-1.27: type-safe and composable, but calls nest and read inside-out —
// the data source ends up buried at the center, reading order opposed to
// execution order.
result := stream.ToSlice( // 5. executed last, written outermost
    stream.Map( // 4. format the top 3
        stream.Limit( // 3. top 3 of the sorted result
            stream.Sorted( // 2. descending sort, forces materialization
                stream.Filter(stream.FromSlice(orders), // 1. the source, read first
                    func(n int) bool { return n > 0 }),
                func(a, b int) int { return b - a },
            ),
            3,
        ),
        func(n int) string { return fmt.Sprintf("$%d", n) },
    ),
)

This library (generic methods):

// Generic methods: reads top-down in pipeline order, element types migrate
// along the chain (int → string), stateful steps slot in seamlessly.
result := stream.FromSlice(orders).
    Filter(func(n int) bool { return n > 0 }).
    Sorted(func(a, b int) int { return b - a }). // stateful: materialize, then sort
    Limit(3). // stateful: top 3 of the sorted result
    Map(func(n int) string { return fmt.Sprintf("$%d", n) }).
    ToSlice()
Style Pros Cons
Plain Go Zero overhead, zero dependencies The stateful steps force two loops plus an in-place sort; laziness, short-circuiting, error propagation, parallelism all hand-rolled; the more stages, the more the loop bodies blur together
Package-level functions (pre-1.27) Type-safe, lazy, composable Nested calls read inside-out, fluent feel lost — the longer the pipeline, the worse
Generic methods (this library) Top-down readability, types flow through the chain; stateful steps slot into the chain seamlessly; laziness/short-circuit/parallelism out of the box Runtime overhead — materialization cost of stateful operators plus dispatch, quantified in Performance below

Readability is half the story; the Performance subsection below quantifies the runtime cost so you can weigh the trade-off for your workload.

Performance

Same pipelines as the style comparison above, each against its hand-written equivalent (BenchmarkTopKVsManual / BenchmarkPipelineVsManual). Both sides play by the same rules: the hand-written version collects into a fresh slice and sorts it in place (unstable pdqsort, same contract as Sorted — the source is never mutated).

Top-K (stateful: Sorted+Limit):

Scale Pipeline Hand-written for Overhead
1e2 ~3.5 μs ~1.5 μs 2.3x
1e4 ~0.17 ms ~0.16 ms 1.1x
1e6 ~17 ms ~7.0 ms 2.4x

Stateless only (Filter+Map+ToSlice):

Scale Pipeline Hand-written for Overhead
1e2 ~2.6 μs ~0.6 μs 4.6x
1e4 ~0.29 ms ~0.17 ms 1.7x
1e6 ~29 ms ~16 ms 1.8x

With the unstable pdqsort as the default, sorting itself becomes cheap and the engine's per-element cost shows through: the remaining gap is dispatch through the sink chain (interface calls + closures, roughly fixed nanoseconds per element), plus per-evaluation setup (~25 small allocations) that dominates at tiny scales. The materialization buffer is a fresh exclusive slice — Sorted/Reverse transform it in place, no extra copy. If you need stable ordering, StableSorted pays the stable-sort cost on both sides (comparable hand-written slices.SortStableFunc code trails by only ~1.2x there, since the sort dominates).

Reproduce with go test -bench . -run '^$' -benchtime 1s (AMD Ryzen 5 7535U, median of 3 runs).

NumberStream

Mirrors Java's primitive streams (IntStream/LongStream) — not for boxing avoidance (Go generics have zero boxing), but for constraint narrowing: NumberStream[N Number] embeds Stream[N], moving the element constraint into the wrapper's own type parameter. This sidesteps the Go 1.27 rule that methods cannot constrain the receiver's existing type parameter, so element-constrained APIs become chainable methods (stream.Range(0, 100).Sum() in one line):

// Narrowed chain: range → filter → sum, no package-level detour
total := stream.Range(1, 101).
    Filter(func(v int) bool { return v%2 == 0 }).
    Sum() // 2550

// Natural-order Sorted/Distinct without comparators or key functions
stream.OfNumber(3, 1, 3, 2, 1).Distinct().Sorted().ToSlice() // [1 2 3]

// Type migration into the narrowed world (Java mapToInt style)
stream.FromSlice(words).MapToNumber(func(s string) int { return len(s) }).Avg()

// Bridging: AsNumber narrows a *Stream; AsStream escapes back
// (for Zip's other side, Chunk/Enumerate, comparator-based Sorted/Min/Max)
stream.Of("a", "b").Zip(stream.Range(1, 10).AsStream(), pair)
stream.AsNumber(stream.Of(1, 2, 3)).Contains(2) // true

Narrowed method surface: element-preserving intermediates (Filter/Peek/TakeWhile/DropWhile/Limit/Skip/Reverse), natural-order ops (Sorted()/StableSorted()/Distinct()), flags/lifecycle (Parallel/Sequential/Unordered/OnClose), and narrowed terminals (Sum()/Avg()/Min()/Max()/Contains()). Non-overridden promoted methods keep Stream semantics: type-migrating operators (Map[U]/Zip/Scan) return *Stream, value terminals (ToSlice/Count/Collect) work directly. Both bridges copy the handle and mark the source consumed — one-shot semantics, second bridge panics.

Performance note: each narrowing entry and element-preserving operator costs one extra handle allocation over the equivalent Stream chain (construction-time only, ~65ns/112B; a depth-4 pure-construction chain measures +5 allocs/+560B; evaluation hot path is identical at n=1e6 — see BenchmarkNumberStreamVsStream). For rebuild-heavy/evaluate-light workloads (tiny inputs, chains rebuilt per request), chain intermediates on *Stream first and narrow with AsNumber just before the terminal.

API Overview

Category APIs
Construction Of OfNonZero FromSlice FromSeq FromChannel FromMap FromFunc Generate Iterate Range RangeClosed Concat Empty
Stateless intermediate Filter Map FlatMap FlatMapSeq Peek TakeWhile DropWhile
Err variants MapErr FilterErr FlatMapErr PeekErr
Stateful intermediate Limit Skip Sorted StableSorted DistinctBy Reverse Scan
Parallelism control Parallel(n) Sequential() Unordered()
Package-level intermediate Distinct Sorted (natural order) Chunk Enumerate WindowSliding
Two-stream Zip
Lifecycle OnClose(f) Close() Cache(s) (replayable factory)
Terminal ForEach ForEachUntil ToSlice ToSeq Count Reduce ReduceOpt Collect First FindAny AnyMatch AllMatch NoneMatch Min Max Err
Collectors (subpackage collector) ToSlice ToSet ToMap ToMapMerge GroupingBy GroupingByDownstream PartitioningBy PartitioningBySlice Teeing Filtering FlatMapping CollectingAndThen MinBy MaxBy Joining Counting Reducing Mapping Summing Averaging Summarizing (SummaryStats)
Numeric constraints stream.Integer/stream.Float/stream.Number (aliases of constraints subpackage)
Package-level aggregation Sum Avg Summary Contains Min Max
Number stream (Task 18) NumberStream[N] (embeds Stream[N]) + narrowing entries Range (returns *NumberStream) OfNumber FromNumberSlice MapToNumber AsNumber/AsStream; narrowed methods Sum() Avg() Min() Max() Contains() Sorted() StableSorted() Distinct()

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 NumberStream[N] narrowing wrapper + Number/cmp.Ordered constraints Boxing avoidance is unnecessary (Go generics have zero boxing); the constraint narrowing value is kept: element-constrained APIs (Sum()/Avg()/Min()/Max()/Contains()/natural-order Sorted()/Distinct()) become chainable methods on NumberStream, mirroring Java's primitive-stream ergonomics
stream.sorted() Sorted (unstable pdqsort) / StableSorted Java's sorted() is always stable; go-stream defaults to the faster unstable sort (aligned with slices.SortFunc) and offers StableSorted when encounter-order preservation matters (aligned with slices.SortStableFunc)
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()
stream.iterator() ToSeq() iter.Seq[T] Outbound adaptation to Go 1.23 range-over-func; consumer break short-circuits the source
Collectors.teeing collector.Teeing One traversal feeds two downstream collectors, then merges both results
Collectors.groupingBy(classifier, downstream) collector.GroupingByDownstream Two-level reduction: group first, then collect each group with a downstream collector (combiner-supported for parallel)
Gatherers.windowSliding(n) WindowSliding(s, n) Full windows only; fewer than n elements produce no output; package-level due to Go 1.27 instantiation-cycle limitation
summaryStatistics() Summary/Summarizing (SummaryStats[N]) Single-pass count/sum/min/max, Avg() derived without a second pass
rangeClosed(a, b) / Stream.ofNullable RangeClosed(a, b) / OfNonZero(xs...) Closed interval; skip zero-value elements (zero covers nil, aligned with cmp.Or terminology)
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.

Roadmap

The project is in the v0.x stage: the API is not yet stable and no compatibility is promised — new features are the priority, but breaking changes may still land between minor releases. Stability guarantees begin with v1.

  • v0.1 (released, tag v0.1.0): 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, automatic fallback to sequential after short-circuit terminals or materializing operators, measured speedup of ~3.3x with 4 shards on CPU-bound workloads); lifecycle & streaming batch — OnClose(f)/Close() resource management, replayable Cache(s) factory, Unordered() streaming merge
  • v0.2 (released, tag v0.2.0): Java 25 parity batch — outbound ToSeq() iter.Seq[T] (range-over-func interop with short-circuit on break), collector composition ecosystem (GroupingByDownstream/PartitioningBy/Teeing/Filtering/FlatMapping/CollectingAndThen/MinBy/MaxBy, combiner-aware for parallel), sliding window WindowSliding, single-pass statistics Summary/Summarizing (SummaryStats), convenience sources RangeClosed/OfNonZero (zero covers nil, aligned with cmp.Or terminology); NumberStream numeric narrowing (NumberStream[N] + MapToNumber/AsNumber bridges); sort semantics split — Sorted (unstable pdqsort, default) vs StableSorted (stable, aligned with slices.SortStableFunc); Collector interface-ization (read-only behavior, regression-benchmarked); new Averaging averaging collector; numeric constraints moved into the constraints subpackage (Summing migrated into collector); Sorted/Reverse transform the materialization buffer in place (saves a full clone)
  • v0.3: scope TBD — the next batch will be scoped from real-world feedback on the v0.2 API surface; suggestions welcome via issues

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)

数值聚合：NumberStream 方法形态（Range 直接收窄，Task 18）。 

package main

import (
	"fmt"

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

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

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).AsStream(), 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）。 单遍求值：和与计数同行累积。收集器形态见 collector.Averaging （需与其它收集器组合时用）；链式方法形态见 (*NumberStream[N]).Avg。

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 判断流中是否含有目标元素（短路）。 与 s.AnyMatch(func(v T) bool { return v == target }) 等价：免写样板闭包， 且 nil 流安全返回 false；any 约束的方法体内无法使用 ==，comparable 只能落在包级。 链式方法形态见 (*NumberStream[N]).Contains（N 为 Number 时）。

func Max 

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

Max 依自然序取最大（空流返回零值与 false）。 免写比较器形态：与 s.Max(cmp.Compare[T]) 等价，方法无法约束 T 故落在包级。 链式方法形态见 (*NumberStream[N]).Max（N 为 Number 时）。

func Min 

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

Min 依自然序取最小（空流返回零值与 false）。 免写比较器形态：与 s.Min(cmp.Compare[T]) 等价，方法无法约束 T 故落在包级。 链式方法形态见 (*NumberStream[N]).Min（N 为 Number 时）。

func Sum 

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

Sum 数值求和。 与 s.Collect(collectors.Summing[N]()) 等价：保留根包一行闭环， 高频终端操作免跨包 import collector 与显式实例化。 链式方法形态见 (*NumberStream[N]).Sum（Task 18，元素约束收窄的流）。

func Summary added in v0.2.0 

func Summary[N Number](s *Stream[N]) collector.SummaryStats[N]

Summary 单遍数值统计：一次遍历同时产出 count/sum/min/max（Avg 经 collector.SummaryStats.Avg() 派生）。与 Sum/Avg 同族的便捷终端， 免 import 子包；需与其它收集器组合（分组统计）时用 collector.Summarizing[N]()。链式方法形态见 (*NumberStream[N]) 经提升 的 Collect(collector.Summarizing[N]())。 空流返回零值统计（Count=0，Avg()=0）。

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 = constraints.Float

Float 约束全部浮点类型。

type Integer 

type Integer = constraints.Integer

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

type KV 

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

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

type Number 

type Number = constraints.Number

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

type NumberStream added in v0.2.0 

type NumberStream[N Number] struct {
	Stream[N]
}

NumberStream 是元素类型收窄为 Number 的流：在 Stream 全部能力之上，提供 依赖数值约束的方法形态终端（Sum/Avg/Min/Max/Contains）与自然序中间操作 （Sorted/StableSorted/Distinct），链式调用不再逃逸到包级函数。

构造：Range（直接收窄）、OfNumber/FromNumberSlice、(*Stream[T]).MapToNumber、 AsNumber 桥接。一次性消费语义与 Stream 完全一致。零值不可用（drive 为 nil）， 一律经构造函数创建。

性能注记：元素保持型中间操作与构造入口相对等价 Stream 版每级多一次句柄 分配（构造期一次性，实测 ~65ns/112B，深度 4 纯构造链合计 +5 allocs/ +560B/+~300ns，见 BenchmarkNumberStreamVsStream），求值热路径与 Stream 零差异（n=1e6 持平）。重构造轻求值的极端场景（每请求重建短链且元素极少） 可先以 *Stream 串联中间操作、末步 AsNumber 收窄后仅接终端。

func AsNumber added in v0.2.0 

func AsNumber[N Number](s *Stream[N]) *NumberStream[N]

AsNumber 把普通流桥接为数值流：收窄约束以解锁 Sum/Avg/Contains/Distinct 等方法形态。桥接即消费——原流 s 被立即标记一次性消费，返回的 NumberStream 持有独立句柄接管管道；二次桥接或继续使用 s 将 panic（fail-fast）。 nil 流返回 nil（与包级便捷函数的容错对齐）。逆操作见 AsStream。

func FromNumberSlice added in v0.2.0 

func FromNumberSlice[N Number](s []N) *NumberStream[N]

FromNumberSlice 基于 slice 构建数值流（FromSlice 的收窄版；零拷贝， 直接引用原切片，求值期间请勿并发修改）。

func OfNumber added in v0.2.0 

func OfNumber[N Number](xs ...N) *NumberStream[N]

OfNumber 以可变参数构建数值流（Of 的收窄版）。

func Range 

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

Range 构建整数区间数值流 [start, stop)（左闭右开，步长 1）。 直接返回 *NumberStream：区间元素必然是数值，就地收窄以解锁 Sum/Avg/Min/Max 等方法形态；需要 *Stream 时经 AsStream 桥接。 （实现随构造函数自 construct.go 迁入，与其余收窄入口同置。）

func RangeClosed added in v0.2.0 

func RangeClosed[I Integer](start, stop I) *NumberStream[I]

RangeClosed 构建整数区间数值流 [start, stop]（左闭右闭，步长 1； start > stop 得空流，对齐 JDK rangeClosed 语义）。与 Range（左闭右开） 并存；直接返回 *NumberStream（与 Range 一致）。 实现基于 rangeSp（stop+1 转为右开）；stop 为类型最大值时 +1 溢出， 以「[start, stop) ++ [stop]」数学等价拆分承接（该极端区间放弃 Sized 特征，遍历语义不变）。

func (*NumberStream[N]) AsStream added in v0.2.0 

func (s *NumberStream[N]) AsStream() *Stream[N]

AsStream 把数值流桥接回普通流：离开数值上下文的显式同型出口（消费本流）。 供 Zip 另一侧、Chunk/Enumerate 等包级函数与被遮蔽的比较器形态 （Sorted/Min/Max 的 cmp 参数版）复用。返回的 *Stream 持有独立句柄接管管道， 本 NumberStream 随即作废。nil 接收者返回 nil。

func (*NumberStream[N]) Avg added in v0.2.0 

func (s *NumberStream[N]) Avg() N

Avg 数值平均（空流返回 0；整数类型按整除）。包级 Avg 的方法形态。

func (*NumberStream[N]) Contains added in v0.2.0 

func (s *NumberStream[N]) Contains(target N) bool

Contains 判断流中是否含有目标元素（包级 Contains 的方法形态； 短路：命中即停止遍历）。N 为 Number（comparable 子集），== 比较在此合法。

func (*NumberStream[N]) Distinct added in v0.2.0 

func (s *NumberStream[N]) Distinct() *NumberStream[N]

Distinct 依据元素自身去重（保留首见，保持遇序；N 为 Number，全部可比较）。 浮点 NaN 互不相等，各 NaN 均保留（同 map 键语义）。本方法为 NumberStream 新增形态；按键去重（键函数任意）用提升的 DistinctBy[K comparable]。

func (*NumberStream[N]) DropWhile added in v0.2.0 

func (s *NumberStream[N]) DropWhile(p func(N) bool) *NumberStream[N]

DropWhile 丢弃首批满足 p 的元素，之后全部放行。

func (*NumberStream[N]) Filter added in v0.2.0 

func (s *NumberStream[N]) Filter(p func(N) bool) *NumberStream[N]

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

func (*NumberStream[N]) Limit added in v0.2.0 

func (s *NumberStream[N]) Limit(n int64) *NumberStream[N]

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

func (*NumberStream[N]) Max added in v0.2.0 

func (s *NumberStream[N]) Max() (N, bool)

Max 依自然序取最大（空流返回零值与 false）。免写比较器形态；本方法 遮蔽 Stream 的比较器版 Max（自定义比较器经 AsStream().Max(cmp)）， 普通 *Stream 的同目的形态为包级函数 Max。

func (*NumberStream[N]) Min added in v0.2.0 

func (s *NumberStream[N]) Min() (N, bool)

Min 依自然序取最小（空流返回零值与 false）。免写比较器形态；本方法 遮蔽 Stream 的比较器版 Min（自定义比较器经 AsStream().Min(cmp)）， 普通 *Stream 的同目的形态为包级函数 Min。

func (*NumberStream[N]) OnClose added in v0.2.0 

func (s *NumberStream[N]) OnClose(f func() error) *NumberStream[N]

OnClose 注册资源清理回调 f：求值结束自动触发一次（幂等，按注册序， 出错记首错可经 Err() 查询）。

func (*NumberStream[N]) Parallel added in v0.2.0 

func (s *NumberStream[N]) Parallel(n int) *NumberStream[N]

Parallel 声明后续求值以最多 n 个分片并行（n <= 1 或不可分源自动串行）。

func (*NumberStream[N]) Peek added in v0.2.0 

func (s *NumberStream[N]) Peek(f func(N)) *NumberStream[N]

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

func (*NumberStream[N]) Reverse added in v0.2.0 

func (s *NumberStream[N]) Reverse() *NumberStream[N]

Reverse 反转元素顺序。

func (*NumberStream[N]) Sequential added in v0.2.0 

func (s *NumberStream[N]) Sequential() *NumberStream[N]

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

func (*NumberStream[N]) Skip added in v0.2.0 

func (s *NumberStream[N]) Skip(n int64) *NumberStream[N]

Skip 跳过前 n 个元素，输出其余（n < 0 panic）。 n == 0 恒等返回自身（不复制句柄、特征位与并行性透传，Task 14 语义）。

func (*NumberStream[N]) Sorted added in v0.2.0 

func (s *NumberStream[N]) Sorted() *NumberStream[N]

Sorted 依自然序（升序）排序，免写比较器（N 为 Number，是 cmp.Ordered 的子集）。不稳定（对齐 slices.SortFunc）。本方法遮蔽 Stream 的比较器版 Sorted 与包级函数 Sorted[T cmp.Ordered]：自定义比较器经 AsStream().Sorted(cmp)，普通 *Stream 用包级 Sorted。

func (*NumberStream[N]) StableSorted added in v0.2.0 

func (s *NumberStream[N]) StableSorted() *NumberStream[N]

StableSorted 依自然序（升序）稳定排序：等值元素保持相遇顺序（对齐 slices.SortStableFunc；纯数值等值即全等，结果与 Sorted 一致，语义上为 需要稳定性的场景预留）。遮蔽 Stream 的比较器版 StableSorted； Stream 侧同目的形态为 s.Stream.StableSorted(cmp)（无包级稳定排序便捷函数）。

func (*NumberStream[N]) Sum added in v0.2.0 

func (s *NumberStream[N]) Sum() N

Sum 数值求和（空流返回 0）。包级 Sum 的方法形态，串行求值语义一致； 并行数值聚合用提升的 Reduce(0, add)（片内折叠、片序合并）。

func (*NumberStream[N]) TakeWhile added in v0.2.0 

func (s *NumberStream[N]) TakeWhile(p func(N) bool) *NumberStream[N]

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

func (*NumberStream[N]) Unordered added in v0.2.0 

func (s *NumberStream[N]) Unordered() *NumberStream[N]

Unordered 声明后续求值不依赖相遇顺序（并行求值按分片完成序流式合并）。

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 约束； 需要按键去重（键函数任意）时用方法版 DistinctBy[K comparable]。 链式方法形态见 (*NumberStream[N]).Distinct（N 为 Number 时）。

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 构建流（零拷贝，直接引用原切片）； 数值收窄版见 FromNumberSlice（number_stream.go）。

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 以可变参数构建流；数值收窄版见 OfNumber（number_stream.go）。

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 OfNonZero added in v0.2.0 

func OfNonZero[T comparable](xs ...T) *Stream[T]

OfNonZero 以可变参数构建流并过滤零值元素（对齐 cmp.Or 的官方术语： nil 是引用类型的零值，zero 涵盖 nil——Java 9 Stream.ofNullable 的 Go 惯用法）。T 须 comparable（零值比较编译期合法；slice/map/func 类型 不可比较，需先经指针或接口包装）。

func Sorted 

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

Sorted 依自然序（cmp.Ordered）排序（不稳定，委托方法 Sorted）。 免写比较器形态：方法版须手写 cmp.Compare[T]，方法无法对 T 追加 cmp.Ordered 约束，故落在包级。需要稳定排序时用 s.StableSorted(cmp.Compare[T])。 链式方法形态见 (*NumberStream[N]).Sorted（N 为 Number 时）。

func WindowSliding added in v0.2.0 

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

WindowSliding 滑动窗口：窗口含 n 个连续元素、逐元素向后滑动，只输出 满窗（长度恰 n），元素少于 n 时无输出（对齐 Java Gatherers windowSliding； 定长不重叠分组用 Chunk）。

包级函数形态：同 Chunk（Go 1.27 泛型方法返回 Stream[[]T] 触发实例化循环）。 有状态单遍（环形缓冲）→ 并行降级（splitN=nil）。n <= 0 panic；nil 流返回 nil。

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 收尾； 收集器 Combiner 返回 nil（不支持并行合并）时自动降级串行。 收集器族见子包 collector（stream/collector）。

func (*Stream[T]) Count 

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

Count 返回元素总数（并行流片内计数、片序求和）。 可组合的收集器形态见 collector.Counting（供 Mapping/GroupingBy 下游计数）。

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 键语义）。 按元素自身 == 去重的免键函数形态见包级函数 Distinct[T comparable]； 元素为数值时另有 (*NumberStream[N]).Distinct()。

func (*Stream[T]) DropWhile 

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

DropWhile 丢弃首批满足 p 的元素，之后全部放行。 有状态单遍（done 门闸）→ 并行降级（splitN=nil）。 数值链形态见 (*NumberStream[N]).DropWhile。

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 的元素。 元素为数值且需要链式调用 Sum/Avg 等收窄终端时，用 NumberStream 的 同名形态（(*NumberStream[N]).Filter，经收窄入口构造）。

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）。 数值链形态见 (*NumberStream[N]).Limit。

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。 类型迁移入数值流的收窄形态见下方 MapToNumber。

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]) MapToNumber added in v0.2.0 

func (s *Stream[T]) MapToNumber[N Number](f func(T) N) *NumberStream[N]

MapToNumber 将每个元素经 f 迁移为数值类型并收窄为数值流（对应 Java mapToInt/mapToDouble：f 的返回类型即窄化后的元素类型，类型迁移 + 约束收窄 一步完成）。1:1 变换：保留 SpSized，仅清 SpSorted/SpDistinct（与 Map 一致）。 实现随 Map 同置本文件；收窄后的链式方法面见 NumberStream（Task 18）。

func (*Stream[T]) Max 

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

Max 返回最大元素（依 cmp）；空流返回 (零值, false)。 免写比较器的自然序形态见包级函数 Max[T cmp.Ordered]（方法无法约束 T）； 元素为数值时另有 (*NumberStream[N]).Max()。

func (*Stream[T]) Min 

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

Min 返回最小元素（依 cmp）；空流返回 (零值, false)。 免写比较器的自然序形态见包级函数 Min[T cmp.Ordered]（方法无法约束 T）； 元素为数值时另有 (*NumberStream[N]).Min()。

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 继承合并后的 回调链）触发，均恰好执行一次。 数值链形态见 (*NumberStream[N]).OnClose。

func (*Stream[T]) Parallel 

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

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

func (*Stream[T]) Peek 

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

Peek 对每个元素施加副作用 f（不改变元素，常用于调试观察）。 并行流下 f 在分片 goroutine 内执行，观察顺序不保证（需保序请用 ForEach）。 数值链形态见 (*NumberStream[N]).Peek。

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 为初值折叠全部元素（并行流片内折叠、片序合并）。 可组合的收集器形态见 collector.Reducing（供 Mapping/GroupingBy 下游折叠）。

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 反转元素顺序。 就地反转物化缓冲（独占切片，非源别名，同 Sorted 的论证）。 数值链形态见 (*NumberStream[N]).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 声明）。 数值链形态见 (*NumberStream[N]).Sequential。

func (*Stream[T]) Skip 

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

Skip 跳过前 n 个元素，输出其余（n == 0 恒等返回原流，不物化、特征位透传； n < 0 panic）。恒等返回时不标记上游 consumed，原流仍可继续链接。 数值链形态见 (*NumberStream[N]).Skip。

func (*Stream[T]) Sorted 

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

Sorted 按比较器 cmp 升序排序（cmp 负/零/正 表示小于/等于/大于）。 不稳定（pdqsort，对齐 slices.SortFunc）：等键元素的相对顺序不保证， 换取更快的默认排序；需要等键保相遇序时用 StableSorted。 免写比较器的自然序形态见包级函数 Sorted[T cmp.Ordered]（方法无法约束 T）； 元素为数值时另有 (*NumberStream[N]).Sorted()。 就地排序物化缓冲：collectingSink 物化的缓冲为本次求值独占的全新切片 （append 构建，非源切片别名），不克隆即排序，省一次全量拷贝； 用户源切片不受影响（回归测试 TestSorted / TestStableSorted 守护）。

func (*Stream[T]) StableSorted added in v0.2.0 

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

StableSorted 按比较器 cmp 升序稳定排序：等键元素保持相遇顺序 （对齐 slices.SortStableFunc，语义同 Java Stream sorted()）。 元素为数值时的免比较器形态见 (*NumberStream[N]).StableSorted()。

func (*Stream[T]) TakeWhile 

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

TakeWhile 保留首批满足 p 的元素，遇到首个不满足即终止（短路）。 数值链形态见 (*NumberStream[N]).TakeWhile。

func (*Stream[T]) ToSeq added in v0.2.0 

func (s *Stream[T]) ToSeq() iter.Seq[T]

ToSeq 把流编译为 Go 1.23 push 迭代器（出站适配，对应 Java stream.iterator()）： 供 range-over-func（for v := range s.ToSeq()）或任何接受 iter.Seq 的 API 消费。 调用即消费本流（终止求值语义）；消费方提前 break 即短路——yield 返回 false 使引擎立即停止推动源；错误即值语义保留（求值后 Err() 可查首错），OnClose 回调链随求值结束照常触发。二次 range 同一 seq 将 panic（流本身一次性， 首遍 range 已将其消费，与全库一次性契约一致）。

func (*Stream[T]) ToSlice 

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

ToSlice 收集全部元素为新切片（并行流按相遇序合并进同一终端）。 可组合的收集器形态见 collector.ToSlice（供 Mapping/GroupingBy 下游收集）。

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 仍按片序聚合（结果不受影响）。 数值链形态见 (*NumberStream[N]).Unordered。

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 接口及其预置实现）。
Package constraints 提供跨包复用的类型约束（零依赖叶子包）。
 Package constraints 提供跨包复用的类型约束（零依赖叶子包）。

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.