feat(stream): 实现流式处理基础功能及并行处理

添加流式处理核心功能，包括Stream结构体、Filter操作和并行处理能力
实现顺序和并行两种处理模式，支持保序和乱序收集
添加单元测试和性能基准测试
配置CI/CD工作流和项目基础文件

Author: xbmlz

.github/workflows/test.yml

@@ -0,0 +1,68 @@
+name: Test
+name: Go CI/CD Pipeline
+
+on:
+  push:
+    branches: [ main, master ]
+  pull_request:
+    branches: [ main, master ]
+
+jobs:
+  test:
+    name: Test on ${{ matrix.os }} with Go ${{ matrix.go-version }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ ubuntu-latest, windows-latest, macos-latest ]
+        go-version: [ '1.21', '1.22', '1.23' ]
+      fail-fast: false
+
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v4
+
+      - name: Set up Go ${{ matrix.go-version }}
+        uses: actions/setup-go@v5
+        with:
+          go-version: ${{ matrix.go-version }}
+          cache-dependency-path: go.sum
+
+      - name: Cache Go modules
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+          key: ${{ matrix.os }}-go-${{ matrix.go-version }}-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ matrix.os }}-go-${{ matrix.go-version }}-
+
+      - name: Install dependencies
+        run: |
+          go mod tidy
+          go mod download
+
+      - name: Run Unit tests with coverage
+        run: |
+          go test -v -race -coverprofile=covprofile -covermode=atomic -coverpkg=./... ./...
+
+      - name: Install golangci-lint
+        if: runner.os == 'Linux'
+        run: go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest
+
+      - name: Run lint (Linux only)
+        if: runner.os == 'Linux'
+        run: golangci-lint run --timeout=5m
+
+      - name: Install goveralls
+        if: runner.os == 'Linux'
+        run: go install github.com/mattn/goveralls@latest
+
+      - name: Send coverage to Coveralls (Linux only)
+        if: runner.os == 'Linux' && github.event_name == 'push' && github.ref == 'refs/heads/main' || github.ref == 'refs/heads/master'
+        env:
+          COVERALLS_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: goveralls -coverprofile=covprofile -service=github -repotoken=${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build
+        run: go build -v ./...

Makefile

@@ -0,0 +1,8 @@
+.PHONY: clean test
+
+test:
+	go test ./...
+	go test -bench=. -benchmem ./...
+
+clean:
+	go clean ./...

README.md

@@ -1,2 +1,9 @@
 # goex
+
 Practical Go extensions for everyday programming: streams, conditions, options, and more.
+
+<a title="Build Status" target="_blank" href="https://github.com/xbmlz/goex/actions/workflows/test.yml"><img src="https://img.shields.io/github/actions/workflow/status/xbmlz/goex/test.yml?style=flat-square"></a>
+<a title="GoDoc" target="_blank" href="https://godoc.org/github.com/xbmlz/goex"><img src="http://img.shields.io/badge/godoc-reference-5272B4.svg?style=flat-square"></a>
+<a title="Go Report Card" target="_blank" href="https://goreportcard.com/report/github.com/xbmlz/goex"><img src="https://goreportcard.com/badge/github.com/xbmlz/goex?style=flat-square"></a>
+<a title="Coverage Status" target="_blank" href="https://coveralls.io/github/xbmlz/goex"><img src="https://img.shields.io/coveralls/github/xbmlz/goex.svg?style=flat-square&color=CC9933"></a>
+<a title="Code Size" target="_blank" href="https://github.com/xbmlz/goex"><img src="https://img.shields.io/github/languages/code-size/xbmlz/goex.svg?style=flat-square"></a>

go.mod

@@ -0,0 +1,3 @@
+module github.com/xbmlz/goex
+
+go 1.24.2

streams/intermediate.go

@@ -0,0 +1,6 @@
+package stream
+
+func (s *Stream[T]) Filter(f func(T) bool) *Stream[T] {
+	s.operations = append(s.operations, FilterOperation[T]{f: f})
+	return s
+}

streams/operations.go

@@ -0,0 +1,20 @@
+package stream
+
+type Operation[T any] interface {
+	Apply(v T) (T, bool)
+	IsOrdered() bool
+}
+
+// FilterOperation filters the stream items.
+type FilterOperation[T any] struct {
+	f func(T) bool
+}
+
+// Apply implements Operation.
+func (op FilterOperation[T]) Apply(v T) (T, bool) {
+	return v, op.f(v)
+}
+
+func (op FilterOperation[T]) IsOrdered() bool {
+	return true
+}

streams/stream.go

@@ -0,0 +1,22 @@
+package stream
+
+type Stream[T any] struct {
+	items      []T
+	operations []Operation[T]
+	parallel   bool
+	ordered    bool
+}
+
+func Of[T any](items []T) *Stream[T] {
+	return &Stream[T]{items: items, ordered: true}
+}
+
+func (s *Stream[T]) Parallel() *Stream[T] {
+	s.parallel = true
+	return s
+}
+
+func (s *Stream[T]) Unordered() *Stream[T] {
+	s.ordered = false
+	return s
+}

streams/stream_benchmark_test.go

@@ -0,0 +1,51 @@
+package stream
+
+import (
+	"testing"
+)
+
+func BenchmarkSequentialFilter(b *testing.B) {
+	// 构造大数据
+	data := make([]int, 1_000_000)
+	for i := 0; i < len(data); i++ {
+		data[i] = i
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = Of(data).
+			Filter(func(x int) bool { return x%2 == 0 }).
+			Collect()
+	}
+}
+
+func BenchmarkParallelFilterOrdered(b *testing.B) {
+	data := make([]int, 1_000_000)
+	for i := 0; i < len(data); i++ {
+		data[i] = i
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = Of(data).
+			Parallel().
+			Filter(func(x int) bool { return x%2 == 0 }).
+			Collect()
+	}
+}
+
+func BenchmarkParallelFilterUnordered(b *testing.B) {
+	data := make([]int, 1_000_000)
+	for i := 0; i < len(data); i++ {
+		data[i] = i
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = Of(data).
+			Parallel().
+			Unordered().
+			Filter(func(x int) bool { return x%2 == 0 }).
+			Collect()
+	}
+}

streams/stream_test.go

@@ -0,0 +1,51 @@
+package stream
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestFilter(t *testing.T) {
+	nums := []int{1, 2, 3, 4, 5, 6}
+
+	// 顺序流
+	expected := []int{2, 4, 6}
+	result := Of(nums).
+		Filter(func(x int) bool { return x%2 == 0 }).
+		Collect()
+	if !reflect.DeepEqual(result, expected) {
+		t.Errorf("Sequential Filter failed. Got %v, want %v", result, expected)
+	}
+
+	// 并行保序流
+	resultParallel := Of(nums).
+		Parallel().
+		Filter(func(x int) bool { return x%2 == 0 }).
+		Collect()
+	if !reflect.DeepEqual(resultParallel, expected) {
+		t.Errorf("Parallel Ordered Filter failed. Got %v, want %v", resultParallel, expected)
+	}
+
+	// 并行乱序流
+	resultUnordered := Of(nums).
+		Parallel().
+		Unordered().
+		Filter(func(x int) bool { return x%2 == 0 }).
+		Collect()
+	// 乱序流只保证元素存在，不保证顺序
+	if len(resultUnordered) != len(expected) {
+		t.Errorf("Parallel Unordered Filter length mismatch. Got %v, want %v", resultUnordered, expected)
+	}
+	for _, v := range expected {
+		found := false
+		for _, u := range resultUnordered {
+			if u == v {
+				found = true
+				break
+			}
+		}
+		if !found {
+			t.Errorf("Parallel Unordered Filter missing element %v", v)
+		}
+	}
+}

streams/terminal.go

@@ -0,0 +1,108 @@
+package stream
+
+import (
+	"runtime"
+	"sync"
+)
+
+// processItem 处理单个元素，应用所有操作
+func processItem[T any](item T, operations []Operation[T]) (T, bool) {
+	res := item
+	ok := true
+	for _, op := range operations {
+		res, ok = op.Apply(res)
+		if !ok {
+			break
+		}
+	}
+	return res, ok
+}
+
+func CollectParallel[T any](s *Stream[T]) []T {
+	if !s.ordered {
+		// 乱序收集
+		var wg sync.WaitGroup
+		ch := make(chan T, len(s.items))
+
+		for _, item := range s.items {
+			wg.Add(1)
+			go func(v T) {
+				defer wg.Done()
+				res, ok := processItem(v, s.operations)
+				if ok {
+					ch <- res
+				}
+			}(item)
+		}
+		wg.Wait()
+		close(ch)
+
+		result := make([]T, 0, len(s.items))
+		for v := range ch {
+			result = append(result, v)
+		}
+		return result
+	}
+	// 保存处理结果和是否保留的信息
+	results := make([]T, len(s.items))
+	keep := make([]bool, len(s.items))
+	var wg sync.WaitGroup
+
+	// 确定工作协程数量 - 不超过CPU核心数或元素数量
+	numWorkers := min(len(s.items), runtime.NumCPU())
+	jobs := make(chan int, len(s.items))
+
+	for w := 0; w < numWorkers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for idx := range jobs {
+				// 处理单个元素
+				res, ok := processItem(s.items[idx], s.operations)
+				// 保存结果并标记为保留
+				if ok {
+					results[idx] = res
+					keep[idx] = true
+				}
+			}
+		}()
+	}
+
+	// 发送任务
+	for i := range s.items {
+		jobs <- i
+	}
+	close(jobs)
+	wg.Wait()
+
+	// 收集保留的结果
+	finalResult := make([]T, 0, len(s.items))
+	for i, ok := range keep {
+		if ok {
+			finalResult = append(finalResult, results[i])
+		}
+	}
+	return finalResult
+}
+
+func (s *Stream[T]) Collect() []T {
+	var result []T
+	if !s.parallel {
+		for _, item := range s.items {
+			res := item
+			ok := true
+			for _, op := range s.operations {
+				res, ok = op.Apply(res)
+				if !ok {
+					break
+				}
+			}
+			if ok {
+				result = append(result, res)
+			}
+		}
+	} else {
+		result = CollectParallel(s)
+	}
+	return result
+}