Go1.18新特性使用Generics泛型進(jìn)行流式處理

更新時(shí)間：2022年06月15日 11:25:50 作者：搖擺的小虎牙

這篇文章主要為大家介紹了Go1.18新特性使用Generics泛型進(jìn)行流式處理詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

Stream 是一個(gè)基于 Go 1.18+ 泛型的流式處理庫, 它支持并行處理流中的數(shù)據(jù). 并行流會(huì)將元素平均劃分多個(gè)的分區(qū), 并創(chuàng)建相同數(shù)量的 goroutine 執(zhí)行, 并且會(huì)保證處理完成后流中元素保持原始順序.

GitHub - xyctruth/stream: A Stream library based on Go 1.18+ Generics (Support Parallel Stream)

安裝

需要安裝 Go 1.18+ 版本

$ go get github.com/xyctruth/stream

在代碼中導(dǎo)入它

import "github.com/xyctruth/stream"

基礎(chǔ)

s := stream.NewSliceByOrdered([]string{"d", "a", "b", "c", "a"}).
    Filter(func(s string) bool { return s != "b" }).
    Map(func(s string) string { return "class_" + s }).
    Sort().
    Distinct().
    ToSlice()
// 需要轉(zhuǎn)換切片元素的類型
s := stream.NewSliceByMapping[int, string, string]([]int{1, 2, 3, 4, 5}).
    Filter(func(v int) bool { return v >3 }).
    Map(func(v int) string { return "mapping_" + strconv.Itoa(v) }).
    Reduce(func(r string, v string) string { return r + v })

類型約束

any 接受任何類型的元素, 所以不能使用 == != > < 比較元素, 導(dǎo)致你不能使用 Sort(), Find()...等函數(shù) ,但是你可以使用 SortFunc(fn), FindFunc(fn)... 代替

type SliceStream[E any] struct {
    slice      []E
}
stream.NewSlice([]int{1, 2, 3, 7, 1})

comparable 接收的類型可以使用 == != 比較元素, 但仍然不能使用 > < 比較元素, 因此你不能使用 Sort(), Min()...等函數(shù) ,但是你可以使用 SortFunc(fn), MinFunc()... 代替

type SliceComparableStream[E comparable] struct {
    SliceStream[E]
}
stream.NewSliceByComparable([]int{1, 2, 3, 7, 1})

constraints.Ordered 接收的類型可以使用 == != > <, 所以可以使用所有的函數(shù)

type SliceOrderedStream[E constraints.Ordered] struct {
    SliceComparableStream[E]
}
stream.NewSliceByOrdered([]int{1, 2, 3, 7, 1})

類型轉(zhuǎn)換

有些時(shí)候我們需要使用 Map ,Reduce 轉(zhuǎn)換切片元素的類型,但是很遺憾目前 Golang 并不支持結(jié)構(gòu)體的方法有額外的類型參數(shù),所有類型參數(shù)必須在結(jié)構(gòu)體中聲明。在 Golang 支持之前我們暫時(shí)使用臨時(shí)方案解決這個(gè)問題。

// SliceMappingStream  Need to convert the type of slice elements.
// - E elements type
// - MapE map elements type
// - ReduceE reduce elements type
type SliceMappingStream[E any, MapE any, ReduceE any] struct {
    SliceStream[E]
}
s := stream.NewSliceByMapping[int, string, string]([]int{1, 2, 3, 4, 5}).
    Filter(func(v int) bool { return v >3 }).
    Map(func(v int) string { return "mapping_" + strconv.Itoa(v) }).
    Reduce(func(r string, v string) string { return r + v })

并行

Parallel 函數(shù)接收一個(gè) goroutines int 參數(shù). 如果 goroutines>1 則開啟并行, 否則關(guān)閉并行, 默認(rèn)流是關(guān)閉并行的。

并行會(huì)將流中的元素平均劃分多個(gè)的分區(qū), 并創(chuàng)建相同數(shù)量的 goroutine 執(zhí)行, 并且會(huì)保證處理完成后流中元素保持原始順序.

s := stream.NewSliceByOrdered([]string{"d", "a", "b", "c", "a"}).
    Parallel(10).
    Filter(func(s string) bool {
    // 一些耗時(shí)操作
    return s != "b"
    }).
    Map(func(s string) string {
    // 一些耗時(shí)操作
    return "class_" + s
    }).
    ForEach(
    func(index int, s string) {
    // 一些耗時(shí)操作
    },
    ).ToSlice()

并行類型

First: 一旦獲得第一個(gè)返回值，并行處理就結(jié)束. For: AllMatch, AnyMatch, FindFunc
ALL: 所有元素都需要并行處理，得到所有返回值，然后并行結(jié)束. For: Map, Filter
Action: 所有元素需要并行處理，不需要返回值. For: ForEach, Action

并行 goroutines

開啟并行 goroutine 數(shù)量在面對(duì) CPU 操作與 IO 操作有著不同的選擇。

一般面對(duì) CPU 操作時(shí) goroutine 數(shù)量不需要設(shè)置大于 CPU 核心數(shù)，而 IO 操作時(shí) goroutine 數(shù)量可以設(shè)置遠(yuǎn)遠(yuǎn)大于 CPU 核心數(shù).

CPU 操作

NewSlice(s).Parallel(goroutines).ForEach(func(i int, v int) {
    sort.Ints(newArray(1000)) //  模擬 CPU 耗時(shí)操作
})

使用6個(gè)cpu核心進(jìn)行基準(zhǔn)測試

go test -run=^$ -benchtime=5s -cpu=6  -bench=^BenchmarkParallelByCPU
goarch: amd64
pkg: github.com/xyctruth/stream
cpu: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
BenchmarkParallelByCPU/no_parallel(0)-6         	     717	   9183119 ns/op
BenchmarkParallelByCPU/goroutines(2)-6          	    1396	   4303113 ns/op
BenchmarkParallelByCPU/goroutines(4)-6          	    2539	   2388197 ns/op
BenchmarkParallelByCPU/goroutines(6)-6          	    2932	   2159407 ns/op
BenchmarkParallelByCPU/goroutines(8)-6          	    2334	   2577405 ns/op
BenchmarkParallelByCPU/goroutines(10)-6         	    2649	   2352926 ns/op

IO 操作

NewSlice(s).Parallel(goroutines).ForEach(func(i int, v int) {
    time.Sleep(time.Millisecond) // 模擬 IO 耗時(shí)操作
})

使用6個(gè)cpu核心進(jìn)行基準(zhǔn)測試

go test -run=^$ -benchtime=5s -cpu=6  -bench=^BenchmarkParallelByIO
goos: darwin
goarch: amd64
pkg: github.com/xyctruth/stream
cpu: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
BenchmarkParallelByIO/no_parallel(0)-6          	      52	 102023558 ns/op
BenchmarkParallelByIO/goroutines(2)-6           	     100	  55807303 ns/op
BenchmarkParallelByIO/goroutines(4)-6           	     214	  27868725 ns/op
BenchmarkParallelByIO/goroutines(6)-6           	     315	  18925789 ns/op
BenchmarkParallelByIO/goroutines(8)-6           	     411	  14439700 ns/op
BenchmarkParallelByIO/goroutines(10)-6          	     537	  11164758 ns/op
BenchmarkParallelByIO/goroutines(50)-6          	    2629	   2310602 ns/op
BenchmarkParallelByIO/goroutines(100)-6         	    5094	   1221887 ns/op

項(xiàng)目地址 https://github.com/xyctruth/stream

以上就是Go1.18新特性使用Generics泛型進(jìn)行流式處理的詳細(xì)內(nèi)容，更多關(guān)于Go1.18 Generics泛型流式處理的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: