Go?通過?Map/Filter/ForEach?等流式?API?高效處理數(shù)據(jù)的思路詳解
用過 Java 的同學(xué)都熟悉 Stream API，那么在 Go 里我們可以用類似的方式處理集合數(shù)據(jù)嗎？本文給大家介紹 go-zero 內(nèi)置的 Stream API，為了幫助理解，函數(shù)主要分為三類：獲取操作、中間處理操作、終結(jié)操作。
什么是流處理
如果有 java 使用經(jīng)驗(yàn)的同學(xué)一定會(huì)對(duì) java8 的 Stream 贊不絕口，極大的提高了們對(duì)于集合類型數(shù)據(jù)的處理能力。
int sum = widgets.stream()
.filter(w -> w.getColor() == RED)
.mapToInt(w -> w.getWeight())
.sum();Stream 能讓我們支持鏈?zhǔn)秸{(diào)用和函數(shù)編程的風(fēng)格來實(shí)現(xiàn)數(shù)據(jù)的處理，看起來數(shù)據(jù)像是在流水線一樣不斷的實(shí)時(shí)流轉(zhuǎn)加工，最終被匯總。Stream 的實(shí)現(xiàn)思想就是將數(shù)據(jù)處理流程抽象成了一個(gè)數(shù)據(jù)流，每次加工后返回一個(gè)新的流供使用。
Stream 功能定義
動(dòng)手寫代碼之前，先想清楚，把需求理清楚是最重要的一步，我們嘗試代入作者的視角來思考整個(gè)組件的實(shí)現(xiàn)流程。首先把底層實(shí)現(xiàn)的邏輯放一下 ,先嘗試從零開始進(jìn)行功能定義 stream 功能。
Stream 的工作流程其實(shí)也屬于生產(chǎn)消費(fèi)者模型，整個(gè)流程跟工廠中的生產(chǎn)流程非常相似，嘗試先定義一下 Stream 的生命周期：
- 創(chuàng)建階段/數(shù)據(jù)獲?。ㄔ希?/li>
- 加工階段/中間處理（流水線加工）
- 匯總階段/終結(jié)操作（最終產(chǎn)品）
下面圍繞 stream 的三個(gè)生命周期開始定義 API：
創(chuàng)建階段
為了創(chuàng)建出數(shù)據(jù)流 stream 這一抽象對(duì)象，可以理解為構(gòu)造器。
我們支持三種方式構(gòu)造 stream，分別是：切片轉(zhuǎn)換，channel 轉(zhuǎn)換，函數(shù)式轉(zhuǎn)換。
注意這個(gè)階段的方法都是普通的公開方法，并不綁定 Stream 對(duì)象。
// 通過可變參數(shù)模式創(chuàng)建 stream
func Just(items ...interface{}) Stream
// 通過 channel 創(chuàng)建 stream
func Range(source <-chan interface{}) Stream
// 通過函數(shù)創(chuàng)建 stream
func From(generate GenerateFunc) Stream
// 拼接 stream
func Concat(s Stream, others ...Stream) Stream加工階段
加工階段需要進(jìn)行的操作往往對(duì)應(yīng)了我們的業(yè)務(wù)邏輯，比如：轉(zhuǎn)換，過濾，去重，排序等等。
這個(gè)階段的 API 屬于 method 需要綁定到 Stream 對(duì)象上。
結(jié)合常用的業(yè)務(wù)場(chǎng)景進(jìn)行如下定義：
// 去除重復(fù)item Distinct(keyFunc KeyFunc) Stream // 按條件過濾item Filter(filterFunc FilterFunc, opts ...Option) Stream // 分組 Group(fn KeyFunc) Stream // 返回前n個(gè)元素 Head(n int64) Stream // 返回后n個(gè)元素 Tail(n int64) Stream // 轉(zhuǎn)換對(duì)象 Map(fn MapFunc, opts ...Option) Stream // 合并item到slice生成新的stream Merge() Stream // 反轉(zhuǎn) Reverse() Stream // 排序 Sort(fn LessFunc) Stream // 作用在每個(gè)item上 Walk(fn WalkFunc, opts ...Option) Stream // 聚合其他Stream Concat(streams ...Stream) Stream
加工階段的處理邏輯都會(huì)返回一個(gè)新的 Stream 對(duì)象，這里有個(gè)基本的實(shí)現(xiàn)范式
匯總階段
匯總階段其實(shí)就是我們想要的處理結(jié)果，比如：是否匹配，統(tǒng)計(jì)數(shù)量，遍歷等等。
// 檢查是否全部匹配 AllMatch(fn PredicateFunc) bool // 檢查是否存在至少一項(xiàng)匹配 AnyMatch(fn PredicateFunc) bool // 檢查全部不匹配 NoneMatch(fn PredicateFunc) bool // 統(tǒng)計(jì)數(shù)量 Count() int // 清空stream Done() // 對(duì)所有元素執(zhí)行操作 ForAll(fn ForAllFunc) // 對(duì)每個(gè)元素執(zhí)行操作 ForEach(fn ForEachFunc)
梳理完組件的需求邊界后，我們對(duì)于即將要實(shí)現(xiàn)的 Stream 有了更清晰的認(rèn)識(shí)。在我的認(rèn)知里面真正的架構(gòu)師對(duì)于需求的把握以及后續(xù)演化能達(dá)到及其精準(zhǔn)的地步，做到這一點(diǎn)離不開對(duì)需求的深入思考以及洞穿需求背后的本質(zhì)。通過代入作者的視角來模擬復(fù)盤整個(gè)項(xiàng)目的構(gòu)建流程，學(xué)習(xí)作者的思維方法論這正是我們學(xué)習(xí)開源項(xiàng)目最大的價(jià)值所在。
好了，我們嘗試定義出完整的 Stream 接口全貌以及函數(shù)。
接口的作用不僅僅是模版作用，還在于利用其抽象能力搭建項(xiàng)目整體的框架而不至于一開始就陷入細(xì)節(jié)，能快速的將我們的思考過程通過接口簡(jiǎn)潔的表達(dá)出來，學(xué)會(huì)養(yǎng)成自頂向下的思維方法從宏觀的角度來觀察整個(gè)系統(tǒng)，一開始就陷入細(xì)節(jié)則很容易拔劍四顧心茫然。。。
rxOptions struct {
unlimitedWorkers bool
workers int
}
Option func(opts *rxOptions)
// key生成器
//item - stream中的元素
KeyFunc func(item interface{}) interface{}
// 過濾函數(shù)
FilterFunc func(item interface{}) bool
// 對(duì)象轉(zhuǎn)換函數(shù)
MapFunc func(intem interface{}) interface{}
// 對(duì)象比較
LessFunc func(a, b interface{}) bool
// 遍歷函數(shù)
WalkFunc func(item interface{}, pip chan<- interface{})
// 匹配函數(shù)
PredicateFunc func(item interface{}) bool
// 對(duì)所有元素執(zhí)行操作
ForAllFunc func(pip <-chan interface{})
// 對(duì)每個(gè)item執(zhí)行操作
ForEachFunc func(item interface{})
// 對(duì)每個(gè)元素并發(fā)執(zhí)行操作
ParallelFunc func(item interface{})
// 對(duì)所有元素執(zhí)行聚合操作
ReduceFunc func(pip <-chan interface{}) (interface{}, error)
// item生成函數(shù)
GenerateFunc func(source <-chan interface{})
Stream interface {
// 去除重復(fù)item
Distinct(keyFunc KeyFunc) Stream
// 按條件過濾item
Filter(filterFunc FilterFunc, opts ...Option) Stream
// 分組
Group(fn KeyFunc) Stream
// 返回前n個(gè)元素
Head(n int64) Stream
// 返回后n個(gè)元素
Tail(n int64) Stream
// 獲取第一個(gè)元素
First() interface{}
// 獲取最后一個(gè)元素
Last() interface{}
// 轉(zhuǎn)換對(duì)象
Map(fn MapFunc, opts ...Option) Stream
// 合并item到slice生成新的stream
Merge() Stream
// 反轉(zhuǎn)
Reverse() Stream
// 排序
Sort(fn LessFunc) Stream
// 作用在每個(gè)item上
Walk(fn WalkFunc, opts ...Option) Stream
// 聚合其他Stream
Concat(streams ...Stream) Stream
// 檢查是否全部匹配
AllMatch(fn PredicateFunc) bool
// 檢查是否存在至少一項(xiàng)匹配
AnyMatch(fn PredicateFunc) bool
// 檢查全部不匹配
NoneMatch(fn PredicateFunc) bool
// 統(tǒng)計(jì)數(shù)量
Count() int
// 清空stream
Done()
// 對(duì)所有元素執(zhí)行操作
ForAll(fn ForAllFunc)
// 對(duì)每個(gè)元素執(zhí)行操作
ForEach(fn ForEachFunc)
}channel() 方法用于獲取 Stream 管道屬性，因?yàn)樵诰唧w實(shí)現(xiàn)時(shí)我們面向的是接口對(duì)象所以暴露一個(gè)私有方法 read 出來。
// 獲取內(nèi)部的數(shù)據(jù)容器channel,內(nèi)部方法
channel() chan interface{}實(shí)現(xiàn)思路
功能定義梳理清楚了，接下來考慮幾個(gè)工程實(shí)現(xiàn)的問題。
如何實(shí)現(xiàn)鏈?zhǔn)秸{(diào)用
鏈?zhǔn)秸{(diào)用，創(chuàng)建對(duì)象用到的 builder 模式可以達(dá)到鏈?zhǔn)秸{(diào)用效果。實(shí)際上 Stream 實(shí)現(xiàn)類似鏈?zhǔn)降男Ч硪彩且粯拥?，每次調(diào)用完后都創(chuàng)建一個(gè)新的 Stream 返回給用戶。
// 去除重復(fù)item Distinct(keyFunc KeyFunc) Stream // 按條件過濾item Filter(filterFunc FilterFunc, opts ...Option) Stream
如何實(shí)現(xiàn)流水線的處理效果
所謂的流水線可以理解為數(shù)據(jù)在 Stream 中的存儲(chǔ)容器，在 go 中我們可以使用 channel 作為數(shù)據(jù)的管道，達(dá)到 Stream 鏈?zhǔn)秸{(diào)用執(zhí)行多個(gè)操作時(shí)異步非阻塞效果。
如何支持并行處理
數(shù)據(jù)加工本質(zhì)上是在處理 channel 中的數(shù)據(jù)，那么要實(shí)現(xiàn)并行處理無非是并行消費(fèi) channel 而已，利用 goroutine 協(xié)程、WaitGroup 機(jī)制可以非常方便的實(shí)現(xiàn)并行處理。
go-zero 實(shí)現(xiàn)
core/fx/stream.go
go-zero 中關(guān)于 Stream 的實(shí)現(xiàn)并沒有定義接口，不過沒關(guān)系底層實(shí)現(xiàn)時(shí)邏輯是一樣的。
為了實(shí)現(xiàn) Stream 接口我們定義一個(gè)內(nèi)部的實(shí)現(xiàn)類，其中 source 為 channel 類型，模擬流水線功能。
Stream struct {
source <-chan interface{}
}創(chuàng)建 API
channel 創(chuàng)建 Range
通過 channel 創(chuàng)建 stream
func Range(source <-chan interface{}) Stream {
return Stream{
source: source,
}
}可變參數(shù)模式創(chuàng)建 Just
通過可變參數(shù)模式創(chuàng)建 stream，channel 寫完后及時(shí) close 是個(gè)好習(xí)慣。
func Just(items ...interface{}) Stream {
source := make(chan interface{}, len(items))
for _, item := range items {
source <- item
}
close(source)
return Range(source)
}函數(shù)創(chuàng)建 From
通過函數(shù)創(chuàng)建 Stream
func From(generate GenerateFunc) Stream {
source := make(chan interface{})
threading.GoSafe(func() {
defer close(source)
generate(source)
})
return Range(source)
}因?yàn)樯婕巴獠總魅氲暮瘮?shù)參數(shù)調(diào)用，執(zhí)行過程并不可用因此需要捕捉運(yùn)行時(shí)異常防止 panic 錯(cuò)誤傳導(dǎo)到上層導(dǎo)致應(yīng)用崩潰。
func Recover(cleanups ...func()) {
for _, cleanup := range cleanups {
cleanup()
}
if r := recover(); r != nil {
logx.ErrorStack(r)
}
}
func RunSafe(fn func()) {
defer rescue.Recover()
fn()
}
func GoSafe(fn func()) {
go RunSafe(fn)
}拼接 Concat
拼接其他 Stream 創(chuàng)建一個(gè)新的 Stream，調(diào)用內(nèi)部 Concat method 方法，后文將會(huì)分析 Concat 的源碼實(shí)現(xiàn)。
func Concat(s Stream, others ...Stream) Stream {
return s.Concat(others...)
}加工 API
去重 Distinct
因?yàn)閭魅氲氖呛瘮?shù)參數(shù)KeyFunc func(item interface{}) interface{}意味著也同時(shí)支持按照業(yè)務(wù)場(chǎng)景自定義去重，本質(zhì)上是利用 KeyFunc 返回的結(jié)果基于 map 實(shí)現(xiàn)去重。
函數(shù)參數(shù)非常強(qiáng)大，能極大的提升靈活性。
func (s Stream) Distinct(keyFunc KeyFunc) Stream {
source := make(chan interface{})
threading.GoSafe(func() {
// channel記得關(guān)閉是個(gè)好習(xí)慣
defer close(source)
keys := make(map[interface{}]lang.PlaceholderType)
for item := range s.source {
// 自定義去重邏輯
key := keyFunc(item)
// 如果key不存在,則將數(shù)據(jù)寫入新的channel
if _, ok := keys[key]; !ok {
source <- item
keys[key] = lang.Placeholder
}
}
})
return Range(source)
}使用案例：
// 1 2 3 4 5
Just(1, 2, 3, 3, 4, 5, 5).Distinct(func(item interface{}) interface{} {
return item
}).ForEach(func(item interface{}) {
t.Log(item)
})
// 1 2 3 4
Just(1, 2, 3, 3, 4, 5, 5).Distinct(func(item interface{}) interface{} {
uid := item.(int)
// 對(duì)大于4的item進(jìn)行特殊去重邏輯,最終只保留一個(gè)>3的item
if uid > 3 {
return 4
}
return item
}).ForEach(func(item interface{}) {
t.Log(item)
})過濾 Filter
通過將過濾邏輯抽象成 FilterFunc，然后分別作用在 item 上根據(jù) FilterFunc 返回的布爾值決定是否寫回新的 channel 中實(shí)現(xiàn)過濾功能，實(shí)際的過濾邏輯委托給了 Walk method。
Option 參數(shù)包含兩個(gè)選項(xiàng)：
- unlimitedWorkers 不限制協(xié)程數(shù)量
- workers 限制協(xié)程數(shù)量
FilterFunc func(item interface{}) bool
func (s Stream) Filter(filterFunc FilterFunc, opts ...Option) Stream {
return s.Walk(func(item interface{}, pip chan<- interface{}) {
if filterFunc(item) {
pip <- item
}
}, opts...)
}使用示例：
func TestInternalStream_Filter(t *testing.T) {
// 保留偶數(shù) 2,4
channel := Just(1, 2, 3, 4, 5).Filter(func(item interface{}) bool {
return item.(int)%2 == 0
}).channel()
for item := range channel {
t.Log(item)
}
}遍歷執(zhí)行 Walk
walk 英文意思是步行，這里的意思是對(duì)每個(gè) item 都執(zhí)行一次 WalkFunc 操作并將結(jié)果寫入到新的 Stream 中。
這里注意一下因?yàn)閮?nèi)部采用了協(xié)程機(jī)制異步執(zhí)行讀取和寫入數(shù)據(jù)所以新的 Stream 中 channel 里面的數(shù)據(jù)順序是隨機(jī)的。
// item-stream中的item元素
// pipe-item符合條件則寫入pipe
WalkFunc func(item interface{}, pipe chan<- interface{})
func (s Stream) Walk(fn WalkFunc, opts ...Option) Stream {
option := buildOptions(opts...)
if option.unlimitedWorkers {
return s.walkUnLimited(fn, option)
}
return s.walkLimited(fn, option)
}
func (s Stream) walkUnLimited(fn WalkFunc, option *rxOptions) Stream {
// 創(chuàng)建帶緩沖區(qū)的channel
// 默認(rèn)為16,channel中元素超過16將會(huì)被阻塞
pipe := make(chan interface{}, defaultWorkers)
go func() {
var wg sync.WaitGroup
for item := range s.source {
// 需要讀取s.source的所有元素
// 這里也說明了為什么channel最后寫完記得完畢
// 如果不關(guān)閉可能導(dǎo)致協(xié)程一直阻塞導(dǎo)致泄漏
// 重要, 不賦值給val是個(gè)典型的并發(fā)陷阱，后面在另一個(gè)goroutine里使用了
val := item
wg.Add(1)
// 安全模式下執(zhí)行函數(shù)
threading.GoSafe(func() {
defer wg.Done()
fn(item, pipe)
})
}
wg.Wait()
close(pipe)
}()
// 返回新的Stream
return Range(pipe)
}
func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
pipe := make(chan interface{}, option.workers)
go func() {
var wg sync.WaitGroup
// 控制協(xié)程數(shù)量
pool := make(chan lang.PlaceholderType, option.workers)
for item := range s.source {
// 重要, 不賦值給val是個(gè)典型的并發(fā)陷阱，后面在另一個(gè)goroutine里使用了
val := item
// 超過協(xié)程限制時(shí)將會(huì)被阻塞
pool <- lang.Placeholder
// 這里也說明了為什么channel最后寫完記得完畢
// 如果不關(guān)閉可能導(dǎo)致協(xié)程一直阻塞導(dǎo)致泄漏
wg.Add(1)
// 安全模式下執(zhí)行函數(shù)
threading.GoSafe(func() {
defer func() {
wg.Done()
//執(zhí)行完成后讀取一次pool釋放一個(gè)協(xié)程位置
<-pool
}()
fn(item, pipe)
})
}
wg.Wait()
close(pipe)
}()
return Range(pipe)
}使用案例：
返回的順序是隨機(jī)的。
func Test_Stream_Walk(t *testing.T) {
// 返回 300,100,200
Just(1, 2, 3).Walk(func(item interface{}, pip chan<- interface{}) {
pip <- item.(int) * 100
}, WithWorkers(3)).ForEach(func(item interface{}) {
t.Log(item)
})
}分組 Group
通過對(duì) item 匹配放入 map 中。
KeyFunc func(item interface{}) interface{}
func (s Stream) Group(fn KeyFunc) Stream {
groups := make(map[interface{}][]interface{})
for item := range s.source {
key := fn(item)
groups[key] = append(groups[key], item)
}
source := make(chan interface{})
go func() {
for _, group := range groups {
source <- group
}
close(source)
}()
return Range(source)
}獲取前 n 個(gè)元素 Head
n 大于實(shí)際數(shù)據(jù)集長(zhǎng)度的話將會(huì)返回全部元素
func (s Stream) Head(n int64) Stream {
if n < 1 {
panic("n must be greather than 1")
}
source := make(chan interface{})
go func() {
for item := range s.source {
n--
// n值可能大于s.source長(zhǎng)度,需要判斷是否>=0
if n >= 0 {
source <- item
}
// let successive method go ASAP even we have more items to skip
// why we don't just break the loop, because if break,
// this former goroutine will block forever, which will cause goroutine leak.
// n==0說明source已經(jīng)寫滿可以進(jìn)行關(guān)閉了
// 既然source已經(jīng)滿足條件了為什么不直接進(jìn)行break跳出循環(huán)呢?
// 作者提到了防止協(xié)程泄漏
// 因?yàn)槊看尾僮髯罱K都會(huì)產(chǎn)生一個(gè)新的Stream,舊的Stream永遠(yuǎn)也不會(huì)被調(diào)用了
if n == 0 {
close(source)
break
}
}
// 上面的循環(huán)跳出來了說明n大于s.source實(shí)際長(zhǎng)度
// 依舊需要顯示關(guān)閉新的source
if n > 0 {
close(source)
}
}()
return Range(source)
}使用示例：
// 返回1,2
func TestInternalStream_Head(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Head(2).channel()
for item := range channel {
t.Log(item)
}
}獲取后 n 個(gè)元素 Tail
這里很有意思，為了確保拿到最后 n 個(gè)元素使用環(huán)形切片 Ring 這個(gè)數(shù)據(jù)結(jié)構(gòu)，先了解一下 Ring 的實(shí)現(xiàn)。
// 環(huán)形切片
type Ring struct {
elements []interface{}
index int
lock sync.Mutex
}
func NewRing(n int) *Ring {
if n < 1 {
panic("n should be greather than 0")
}
return &Ring{
elements: make([]interface{}, n),
}
}
// 添加元素
func (r *Ring) Add(v interface{}) {
r.lock.Lock()
defer r.lock.Unlock()
// 將元素寫入切片指定位置
// 這里的取余實(shí)現(xiàn)了循環(huán)寫效果
r.elements[r.index%len(r.elements)] = v
// 更新下次寫入位置
r.index++
}
// 獲取全部元素
// 讀取順序保持與寫入順序一致
func (r *Ring) Take() []interface{} {
r.lock.Lock()
defer r.lock.Unlock()
var size int
var start int
// 當(dāng)出現(xiàn)循環(huán)寫的情況時(shí)
// 開始讀取位置需要通過去余實(shí)現(xiàn),因?yàn)槲覀兿Ｍx取出來的順序與寫入順序一致
if r.index > len(r.elements) {
size = len(r.elements)
// 因?yàn)槌霈F(xiàn)循環(huán)寫情況,當(dāng)前寫入位置index開始為最舊的數(shù)據(jù)
start = r.index % len(r.elements)
} else {
size = r.index
}
elements := make([]interface{}, size)
for i := 0; i < size; i++ {
// 取余實(shí)現(xiàn)環(huán)形讀取,讀取順序保持與寫入順序一致
elements[i] = r.elements[(start+i)%len(r.elements)]
}
return elements
}總結(jié)一下環(huán)形切片的優(yōu)點(diǎn)：
- 支持自動(dòng)滾動(dòng)更新
- 節(jié)省內(nèi)存
環(huán)形切片能實(shí)現(xiàn)固定容量滿的情況下舊數(shù)據(jù)不斷被新數(shù)據(jù)覆蓋，由于這個(gè)特性可以用于讀取 channel 后 n 個(gè)元素。
func (s Stream) Tail(n int64) Stream {
if n < 1 {
panic("n must be greather than 1")
}
source := make(chan interface{})
go func() {
ring := collection.NewRing(int(n))
// 讀取全部元素，如果數(shù)量>n環(huán)形切片能實(shí)現(xiàn)新數(shù)據(jù)覆蓋舊數(shù)據(jù)
// 保證獲取到的一定最后n個(gè)元素
for item := range s.source {
ring.Add(item)
}
for _, item := range ring.Take() {
source <- item
}
close(source)
}()
return Range(source)
}那么為什么不直接使用 len(source) 長(zhǎng)度的切片呢?
答案是節(jié)省內(nèi)存。凡是涉及到環(huán)形類型的數(shù)據(jù)結(jié)構(gòu)時(shí)都具備一個(gè)優(yōu)點(diǎn)那就省內(nèi)存，能做到按需分配資源。
使用示例：
func TestInternalStream_Tail(t *testing.T) {
// 4,5
channel := Just(1, 2, 3, 4, 5).Tail(2).channel()
for item := range channel {
t.Log(item)
}
// 1,2,3,4,5
channel2 := Just(1, 2, 3, 4, 5).Tail(6).channel()
for item := range channel2 {
t.Log(item)
}
}元素轉(zhuǎn)換Map
元素轉(zhuǎn)換，內(nèi)部由協(xié)程完成轉(zhuǎn)換操作，注意輸出channel并不保證按原序輸出。
MapFunc func(intem interface{}) interface{}
func (s Stream) Map(fn MapFunc, opts ...Option) Stream {
return s.Walk(func(item interface{}, pip chan<- interface{}) {
pip <- fn(item)
}, opts...)
}使用示例：
func TestInternalStream_Map(t *testing.T) {
channel := Just(1, 2, 3, 4, 5, 2, 2, 2, 2, 2, 2).Map(func(item interface{}) interface{} {
return item.(int) * 10
}).channel()
for item := range channel {
t.Log(item)
}
}合并 Merge
實(shí)現(xiàn)比較簡(jiǎn)單，我考慮了很久沒想到有什么場(chǎng)景適合這個(gè)方法。
func (s Stream) Merge() Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
source := make(chan interface{}, 1)
source <- items
return Range(source)
}反轉(zhuǎn) Reverse
反轉(zhuǎn) channel 中的元素。反轉(zhuǎn)算法流程是：
- 找到中間節(jié)點(diǎn)
- 節(jié)點(diǎn)兩邊開始兩兩交換
注意一下為什么獲取 s.source 時(shí)用切片來接收呢? 切片會(huì)自動(dòng)擴(kuò)容，用數(shù)組不是更好嗎?
其實(shí)這里是不能用數(shù)組的，因?yàn)椴恢?Stream 寫入 source 的操作往往是在協(xié)程異步寫入的，每個(gè) Stream 中的 channel 都可能在動(dòng)態(tài)變化，用流水線來比喻 Stream 工作流程的確非常形象。
func (s Stream) Reverse() Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
for i := len(items)/2 - 1; i >= 0; i-- {
opp := len(items) - 1 - i
items[i], items[opp] = items[opp], items[i]
}
return Just(items...)
}使用示例：
func TestInternalStream_Reverse(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Reverse().channel()
for item := range channel {
t.Log(item)
}
}排序 Sort
內(nèi)網(wǎng)調(diào)用 slice 官方包的排序方案，傳入比較函數(shù)實(shí)現(xiàn)比較邏輯即可。
func (s Stream) Sort(fn LessFunc) Stream {
var items []interface{}
for item := range s.source {
items = append(items, item)
}
sort.Slice(items, func(i, j int) bool {
return fn(i, j)
})
return Just(items...)
}使用示例：
// 5,4,3,2,1
func TestInternalStream_Sort(t *testing.T) {
channel := Just(1, 2, 3, 4, 5).Sort(func(a, b interface{}) bool {
return a.(int) > b.(int)
}).channel()
for item := range channel {
t.Log(item)
}
}拼接 Concat
func (s Stream) Concat(steams ...Stream) Stream {
// 創(chuàng)建新的無緩沖channel
source := make(chan interface{})
go func() {
// 創(chuàng)建一個(gè)waiGroup對(duì)象
group := threading.NewRoutineGroup()
// 異步從原channel讀取數(shù)據(jù)
group.Run(func() {
for item := range s.source {
source <- item
}
})
// 異步讀取待拼接Stream的channel數(shù)據(jù)
for _, stream := range steams {
// 每個(gè)Stream開啟一個(gè)協(xié)程
group.Run(func() {
for item := range stream.channel() {
source <- item
}
})
}
// 阻塞等待讀取完成
group.Wait()
close(source)
}()
// 返回新的Stream
return Range(source)
}匯總 API
全部匹配 AllMatch
func (s Stream) AllMatch(fn PredicateFunc) bool {
for item := range s.source {
if !fn(item) {
// 需要排空 s.source，否則前面的goroutine可能阻塞
go drain(s.source)
return false
}
}
return true
}任意匹配 AnyMatch
func (s Stream) AnyMatch(fn PredicateFunc) bool {
for item := range s.source {
if fn(item) {
// 需要排空 s.source，否則前面的goroutine可能阻塞
go drain(s.source)
return true
}
}
return false
}一個(gè)也不匹配 NoneMatch
func (s Stream) NoneMatch(fn func(item interface{}) bool) bool {
for item := range s.source {
if fn(item) {
// 需要排空 s.source，否則前面的goroutine可能阻塞
go drain(s.source)
return false
}
}
return true
}數(shù)量統(tǒng)計(jì) Count
func (s Stream) Count() int {
var count int
for range s.source {
count++
}
return count
}清空 Done
func (s Stream) Done() {
// 排空 channel，防止 goroutine 阻塞泄露
drain(s.source)
}迭代全部元素 ForAll
func (s Stream) ForAll(fn ForAllFunc) {
fn(s.source)
}迭代每個(gè)元素 ForEach
func (s Stream) ForAll(fn ForAllFunc) {
fn(s.source)
}小結(jié)
至此 Stream 組件就全部實(shí)現(xiàn)完了，核心邏輯是利用 channel 當(dāng)做管道，數(shù)據(jù)當(dāng)做水流，不斷的用協(xié)程接收/寫入數(shù)據(jù)到 channel 中達(dá)到異步非阻塞的效果。
回到開篇提到的問題，未動(dòng)手前想要實(shí)現(xiàn)一個(gè) stream 難度似乎非常大，很難想象在 go 中 300 多行的代碼就能實(shí)現(xiàn)如此強(qiáng)大的組件。
實(shí)現(xiàn)高效的基礎(chǔ)來源三個(gè)語言特性：
- channel
- 協(xié)程
- 函數(shù)式編程
參考資料
項(xiàng)目地址
https://github.com/zeromicro/go-zero
到此這篇關(guān)于Go 通過 Map/Filter/ForEach 等流式 API 高效處理數(shù)據(jù)的文章就介紹到這了,更多相關(guān)go 流式 API 處理數(shù)據(jù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
基于go語言實(shí)現(xiàn)圖片驗(yàn)證碼的代碼示例
這篇文章主要為大家詳細(xì)介紹了基于go語言實(shí)現(xiàn)圖片驗(yàn)證碼的代碼示例,文中的示例代碼簡(jiǎn)潔易懂,具有一定的借鑒價(jià)值,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-10-10
相信小伙伴都遇到過以下的循環(huán)變量的問題，本文主要介紹了go中for?range的坑以及解決方案，具有一定的參考價(jià)值，感興趣的可以了解一下2024-01-01
這篇文章主要為大家介紹了Go語言基礎(chǔ)關(guān)于go接口的用法示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，在日升職加薪2021-11-11
go并發(fā)實(shí)現(xiàn)素?cái)?shù)篩的代碼
這篇文章主要介紹了go并發(fā)實(shí)現(xiàn)素?cái)?shù)篩的代碼,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2021-03-03
Go語言題解LeetCode1266訪問所有點(diǎn)的最小時(shí)間示例
這篇文章主要為大家介紹了Go語言題解LeetCode1266訪問所有點(diǎn)的最小時(shí)間示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-01-01
Go語言利用heap實(shí)現(xiàn)優(yōu)先級(jí)隊(duì)列
這篇文章主要為大家詳細(xì)介紹了Go語言中heap的使用以及如何利用heap實(shí)現(xiàn)優(yōu)先級(jí)隊(duì)列的相關(guān)資料，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-05-05
使用Go語言實(shí)現(xiàn)Yaml編碼和解碼的方法詳解
在這篇文章中,我們將介紹如何使用Go語言編寫代碼來實(shí)現(xiàn)Yaml編碼和解碼,文中有詳細(xì)的代碼示例供大家參考,對(duì)大家的學(xué)習(xí)和工作有一定的幫助,需要的朋友可以參考下2023-11-11
詳解如何在Go中循環(huán)中使用Defer關(guān)鍵字示例詳解
這篇文章主要為大家介紹了詳解如何在Go中循環(huán)中使用Defer關(guān)鍵字示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-09-09
Go語言k8s?kubernetes使用leader?election實(shí)現(xiàn)選舉
這篇文章主要為大家介紹了Go語言?k8s?kubernetes?使用leader?election選舉，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-10-10
這篇文章主要為大家介紹了詳解如何使用Golang擴(kuò)展Envoy實(shí)現(xiàn)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-06-06