什么是流處理

如果有 java 使用經(jīng)驗(yàn)的同學(xué)一定會對 java8 的 Stream 贊不絕口绢涡，極大的提高了們對于集合類型數(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í)時流轉(zhuǎn)加工杆勇，最終被匯總。Stream 的實(shí)現(xiàn)思想就是將數(shù)據(jù)處理流程抽象成了一個數(shù)據(jù)流，每次加工后返回一個新的流供使用趟畏。

Stream 功能定義

動手寫代碼之前，先想清楚滩租，把需求理清楚是最重要的一步拱镐，我們嘗試代入作者的視角來思考整個組件的實(shí)現(xiàn)流程。首先把底層實(shí)現(xiàn)的邏輯放一下 ,先嘗試從零開始進(jìn)行功能定義 stream 功能持际。

Stream 的工作流程其實(shí)也屬于生產(chǎn)消費(fèi)者模型沃琅，整個流程跟工廠中的生產(chǎn)流程非常相似，嘗試先定義一下 Stream 的生命周期：

1. 創(chuàng)建階段/數(shù)據(jù)獲戎┯（原料）
2. 加工階段/中間處理（流水線加工）
3. 匯總階段/終結(jié)操作（最終產(chǎn)品）

下面圍繞 stream 的三個生命周期開始定義 API：

創(chuàng)建階段

為了創(chuàng)建出數(shù)據(jù)流 stream 這一抽象對象益眉，可以理解為構(gòu)造器。

我們支持三種方式構(gòu)造 stream姥份，分別是：切片轉(zhuǎn)換郭脂，channel 轉(zhuǎn)換，函數(shù)式轉(zhuǎn)換澈歉。

注意這個階段的方法都是普通的公開方法展鸡，并不綁定 Stream 對象。

// 通過可變參數(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)行的操作往往對應(yīng)了我們的業(yè)務(wù)邏輯埃难，比如：轉(zhuǎn)換莹弊，過濾涤久，去重，排序等等忍弛。

這個階段的 API 屬于 method 需要綁定到 Stream 對象上响迂。

結(jié)合常用的業(yè)務(wù)場景進(jìn)行如下定義：

// 去除重復(fù)item
Distinct(keyFunc KeyFunc) Stream
// 按條件過濾item
Filter(filterFunc FilterFunc, opts ...Option) Stream
// 分組
Group(fn KeyFunc) Stream
// 返回前n個元素
Head(n int64) Stream
// 返回后n個元素
Tail(n int64) Stream
// 轉(zhuǎn)換對象
Map(fn MapFunc, opts ...Option) Stream
// 合并item到slice生成新的stream
Merge() Stream
// 反轉(zhuǎn)
Reverse() Stream
// 排序
Sort(fn LessFunc) Stream
// 作用在每個item上
Walk(fn WalkFunc, opts ...Option) Stream
// 聚合其他Stream
Concat(streams ...Stream) Stream

加工階段的處理邏輯都會返回一個新的 Stream 對象，這里有個基本的實(shí)現(xiàn)范式

image

匯總階段

匯總階段其實(shí)就是我們想要的處理結(jié)果细疚，比如：是否匹配蔗彤，統(tǒng)計數(shù)量，遍歷等等疯兼。

// 檢查是否全部匹配
AllMatch(fn PredicateFunc) bool
// 檢查是否存在至少一項(xiàng)匹配
AnyMatch(fn PredicateFunc) bool
// 檢查全部不匹配
NoneMatch(fn PredicateFunc) bool
// 統(tǒng)計數(shù)量
Count() int
// 清空stream
Done()
// 對所有元素執(zhí)行操作
ForAll(fn ForAllFunc)
// 對每個元素執(zhí)行操作
ForEach(fn ForEachFunc)

梳理完組件的需求邊界后然遏，我們對于即將要實(shí)現(xiàn)的 Stream 有了更清晰的認(rèn)識。在我的認(rèn)知里面真正的架構(gòu)師對于需求的把握以及后續(xù)演化能達(dá)到及其精準(zhǔn)的地步吧彪，做到這一點(diǎn)離不開對需求的深入思考以及洞穿需求背后的本質(zhì)啦鸣。通過代入作者的視角來模擬復(fù)盤整個項(xiàng)目的構(gòu)建流程，學(xué)習(xí)作者的思維方法論這正是我們學(xué)習(xí)開源項(xiàng)目最大的價值所在来氧。

好了诫给，我們嘗試定義出完整的 Stream 接口全貌以及函數(shù)。

接口的作用不僅僅是模版作用啦扬，還在于利用其抽象能力搭建項(xiàng)目整體的框架而不至于一開始就陷入細(xì)節(jié)中狂，能快速的將我們的思考過程通過接口簡潔的表達(dá)出來，學(xué)會養(yǎng)成自頂向下的思維方法從宏觀的角度來觀察整個系統(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
// 對象轉(zhuǎn)換函數(shù)
MapFunc func(intem interface{}) interface{}
// 對象比較
LessFunc func(a, b interface{}) bool
// 遍歷函數(shù)
WalkFunc func(item interface{}, pip chan<- interface{})
// 匹配函數(shù)
PredicateFunc func(item interface{}) bool
// 對所有元素執(zhí)行操作
ForAllFunc func(pip <-chan interface{})
// 對每個item執(zhí)行操作
ForEachFunc func(item interface{})
// 對每個元素并發(fā)執(zhí)行操作
ParallelFunc func(item interface{})
// 對所有元素執(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個元素
  Head(n int64) Stream
  // 返回后n個元素
  Tail(n int64) Stream
  // 獲取第一個元素
  First() interface{}
  // 獲取最后一個元素
  Last() interface{}
  // 轉(zhuǎn)換對象
  Map(fn MapFunc, opts ...Option) Stream
  // 合并item到slice生成新的stream
  Merge() Stream
  // 反轉(zhuǎn)
  Reverse() Stream
  // 排序
  Sort(fn LessFunc) Stream
  // 作用在每個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)計數(shù)量
  Count() int
  // 清空stream
  Done()
  // 對所有元素執(zhí)行操作
  ForAll(fn ForAllFunc)
  // 對每個元素執(zhí)行操作
  ForEach(fn ForEachFunc)
}

channel() 方法用于獲取 Stream 管道屬性勋又，因?yàn)樵诰唧w實(shí)現(xiàn)時我們面向的是接口對象所以暴露一個私有方法 read 出來。

// 獲取內(nèi)部的數(shù)據(jù)容器channel,內(nèi)部方法
channel() chan interface{}

實(shí)現(xiàn)思路

功能定義梳理清楚了换帜，接下來考慮幾個工程實(shí)現(xiàn)的問題楔壤。

如何實(shí)現(xiàn)鏈?zhǔn)秸{(diào)用

鏈?zhǔn)秸{(diào)用，創(chuàng)建對象用到的 builder 模式可以達(dá)到鏈?zhǔn)秸{(diào)用效果惯驼。實(shí)際上 Stream 實(shí)現(xiàn)類似鏈?zhǔn)降男Ч硪彩且粯拥亩紫看握{(diào)用完后都創(chuàng)建一個新的 Stream 返回給用戶。

// 去除重復(fù)item
Distinct(keyFunc KeyFunc) Stream
// 按條件過濾item
Filter(filterFunc FilterFunc, opts ...Option) Stream

如何實(shí)現(xiàn)流水線的處理效果

所謂的流水線可以理解為數(shù)據(jù)在 Stream 中的存儲容器祟牲，在 go 中我們可以使用 channel 作為數(shù)據(jù)的管道隙畜，達(dá)到 Stream 鏈?zhǔn)秸{(diào)用執(zhí)行多個操作時異步非阻塞效果。

如何支持并行處理

數(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í)現(xiàn) Stream 接口我們定義一個內(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 寫完后及時 close 是個好習(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)行時異常防止 panic 錯誤傳導(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)建一個新的 Stream躁愿，調(diào)用內(nèi)部 Concat method 方法，后文將會分析 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{}意味著也同時支持按照業(yè)務(wù)場景自定義去重彤钟，本質(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)閉是個好習(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)
  // 對大于4的item進(jìn)行特殊去重邏輯,最終只保留一個>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ù)包含兩個選項(xiàng)：

1. unlimitedWorkers 不限制協(xié)程數(shù)量
2. 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 英文意思是步行，這里的意思是對每個 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將會被阻塞
  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是個典型的并發(fā)陷阱，后面在另一個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是個典型的并發(fā)陷阱沸枯，后面在另一個goroutine里使用了
      val := item
      // 超過協(xié)程限制時將會被阻塞
      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釋放一個協(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

通過對 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 個元素 Head

n 大于實(shí)際數(shù)據(jù)集長度的話將會返回全部元素

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長度,需要判斷是否>=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都會產(chǎn)生一個新的Stream,舊的Stream永遠(yuǎn)也不會被調(diào)用了
      if n == 0 {
        close(source)
        break
      }
    }
    // 上面的循環(huán)跳出來了說明n大于s.source實(shí)際長度
    // 依舊需要顯示關(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 個元素 Tail

這里很有意思绑榴，為了確保拿到最后 n 個元素使用環(huán)形切片 Ring 這個數(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í)現(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)：

• 支持自動滾動更新
• 節(jié)省內(nèi)存

環(huán)形切片能實(shí)現(xiàn)固定容量滿的情況下舊數(shù)據(jù)不斷被新數(shù)據(jù)覆蓋翔怎，由于這個特性可以用于讀取 channel 后 n 個元素缔逛。

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個元素
    for item := range s.source {
      ring.Add(item)
    }
    for _, item := range ring.Take() {
      source <- item
    }
    close(source)
  }()
  return Range(source)
}

那么為什么不直接使用 len(source) 長度的切片呢?

答案是節(jié)省內(nèi)存姓惑。凡是涉及到環(huán)形類型的數(shù)據(jù)結(jié)構(gòu)時都具備一個優(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)比較簡單唯沮，我考慮了很久沒想到有什么場景適合這個方法脖旱。

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 時用切片來接收呢? 切片會自動擴(kuò)容，用數(shù)組不是更好嗎?

其實(shí)這里是不能用數(shù)組的萌庆，因?yàn)椴恢?Stream 寫入 source 的操作往往是在協(xié)程異步寫入的溶褪，每個 Stream 中的 channel 都可能在動態(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)建一個waiGroup對象
    group := threading.NewRoutineGroup()
    // 異步從原channel讀取數(shù)據(jù)
    group.Run(func() {
      for item := range s.source {
        source <- item
      }
    })
    // 異步讀取待拼接Stream的channel數(shù)據(jù)
    for _, stream := range steams {
      // 每個Stream開啟一個協(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
}

一個也不匹配 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)計 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)
}

迭代每個元素 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á)到異步非阻塞的效果芬萍。

回到開篇提到的問題，未動手前想要實(shí)現(xiàn)一個 stream 難度似乎非常大搔啊，很難想象在 go 中 300 多行的代碼就能實(shí)現(xiàn)如此強(qiáng)大的組件柬祠。

實(shí)現(xiàn)高效的基礎(chǔ)來源三個語言特性：

• channel
• 協(xié)程
• 函數(shù)式編程

參考資料

pipeline模式

切片反轉(zhuǎn)算法

項(xiàng)目地址

https://github.com/zeromicro/go-zero

歡迎使用 go-zero 并 star 支持我們！

微信交流群

關(guān)注『微服務(wù)實(shí)踐』公眾號并點(diǎn)擊 交流群 獲取社區(qū)群二維碼坯癣。