如何把Golang的channel用的如nodejs的stream同樣絲滑

若是讓我和別人說說Golang有什麼特色，我首先想到不必定是goroutine，但必定會是channel。

由於Channel的存在，是讓Goroutine威力加成的利器。

若是用一句話來解釋channel的做用，我會說

Chanel是一個管道，它會讓數據流動起來。

++那麼如何理解這個讓數據流程起來呢？++

假如說你須要對100次請求，作兩種比較耗時的操做，而後再統計加權結果，還須要儘量的併發來提升性能。示例代碼以下：

var multipleChan = make(chan int, 4)
var minusChan = make(chan int, 4)
var harvestChan = make(chan int, 4)

defer close(multipleChan)
defer close(minusChan)
defer close(harvestChan)

go func() {
	for i:=1;i<=100;i++{
		multipleChan <- i
	}
}()

for i:=0; i<4; i++{
	go func() {
		for data := range multipleChan {
			minusChan <- data * 2
			time.Sleep(10* time.Millisecond)
		}
	}()

	go func() {
		for data := range minusChan {
			harvestChan <- data - 1
			time.Sleep(10* time.Millisecond)
		}
	}()
}

var sum = 0
var index = 0
for data := range harvestChan{
	sum += data
	index++
	if index == 100{
		break
	}
}

fmt.Println(sum)
複製代碼

不要笑這段代碼簡單，若是考慮到錯誤處理的狀況，那仍是有些複雜的。好比，某個環節是遇到錯誤能夠忽略，某個環節是遇到要終止全部操做；再加上，有時只關心第一個知足條件的返回值，還須要超時處理。

寫一遍也許還能夠，要是不少地方都要這樣寫，那真是頭大>_<!!!

重複的代碼是萬惡之源，Don't repeat yourself是成爲優秀工程師的第一步。

因而，channelx這個庫誕生了！

使用了這個庫，實現上述一樣的功能，代碼是這樣子的~~

var sum = 0

NewChannelStream(func(seedChan chan<- Result, quitChannel chan struct{}) {
	for i:=1; i<=100;i++{
		seedChan <- Result{Data:i}
	}
	close(seedChan) //記得關閉哦~~~
}).Pipe(func(result Result) Result {
	return Result{Data: result.Data.(int) * 2}
}).Pipe(func(result Result) Result {
	return Result{Data: result.Data.(int) - 1}
}).Harvest(func(result Result) {
	sum += result.Data.(int)
})

fmt.Println(sum)
複製代碼

我喜歡鏈式風格，因此寫成這個樣子，你也能夠拆開來寫的。

但重點是代碼這樣寫起來是否是很絲滑，如寫nodejs stream的感受呢，嘻嘻~~

除了Pipe->Harvest的組合，還能夠實現Pipe->Race, Pipe->Drain, Pipe->Cancel等操做的組合。

這些複雜的例子，均可以參照stream_test.go文件中的單元測試來實現，就不一一貼代碼出來了哈。

那麼，這個stream又是如何實現的呢？核心就在NewChannelStream和Pipe這個兩個函數裏。

func NewChannelStream(seedFunc SeedFunc, optionFuncs ...OptionFunc) *ChannelStream {
	cs := &ChannelStream{
		workers:     runtime.NumCPU(),
		optionFuncs: optionFuncs,
	}

	for _, of := range optionFuncs {
		of(cs)
	}

	if cs.quitChan == nil {
		cs.quitChan = make(chan struct{})
	}

	cs.dataChannel = make(chan Item, cs.workers)

	go func() {
		inputChan := make(chan Item)

		go seedFunc(inputChan, cs.quitChan)

	loop:
		for {
			select {
			case <-cs.quitChan:
				break loop

			case res, ok := <-inputChan:
				if !ok {
					break loop
				}

				select {
				case <-cs.quitChan:
					break loop
				default:
				}

				if res.Err != nil {
					cs.errors = append(cs.errors, res.Err)
				}

				if !cs.hasError && res.Err != nil {
					cs.hasError = true
					cs.dataChannel <- res
					if cs.ape == stop {
						cs.Cancel()
					}
					continue
				}

				if cs.hasError && cs.ape == stop {
					continue
				}

				cs.dataChannel <- res
			}
		}

		safeCloseChannel(cs.dataChannel)

	}()

	return cs
}

func (p *ChannelStream) Pipe(dataPipeFunc PipeFunc, optionFuncs ...OptionFunc) *ChannelStream {
	seedFunc := func(dataPipeChannel chan<- Item, quitChannel chan struct{}) {
		wg := &sync.WaitGroup{}
		wg.Add(p.workers)
		for i := 0; i < p.workers; i++ {
			go func() {
				defer wg.Done()
			loop:
				for {
					select {
					case <-quitChannel:
						break loop
					case data, ok := <-p.dataChannel:
						if !ok {
							break loop
						}

						select {
						case <-quitChannel:
							break loop
						default:
						}

						dataPipeChannel <- dataPipeFunc(data)
					}
				}
			}()
		}

		go func() {
			wg.Wait()
			safeCloseChannel(dataPipeChannel)
		}()
	}

	mergeOptionFuncs := make([]OptionFunc, len(p.optionFuncs)+len(optionFuncs)+1)
	copy(mergeOptionFuncs[0:len(p.optionFuncs)], p.optionFuncs)
	copy(mergeOptionFuncs[len(p.optionFuncs):], optionFuncs)
	mergeOptionFuncs[len(p.optionFuncs)+len(optionFuncs)] = passByQuitChan(p.quitChan) //這行保證了整個stream中有一個惟一的quitChan

	return NewChannelStream(seedFunc, mergeOptionFuncs...)
}

複製代碼

代碼看着多，刨除初始化的代碼、錯誤處理和退出處理的代碼，核心仍是經過channel的數據流動。

首先，NewChannelStream中會新建一個inputChan傳入seedFunc，而後數據會經過seedChan(即inputChan)，傳到dataChannel。

而後，當調用Pipe的時候，Pipe函數會本身建立一個seedFunc從上一個channelStream的dataChannel傳到dataPipeChannel中。這個Pipe中的seedFunc又會傳入NewChannelStream中，產生一個新channelStream對象，這時在新的channelStream中，inputChan即Pipe中的dataPipeChannel，整個數據流就這樣串起來了，過程以下：

inputChan(seedChan)->dataChannel->inputChan(dataPipeChannel)->dataChannel->....

分析過源碼，再來看使用ChannelStream的例子和直接用Channel的例子，兩個dataChannel分別對應的是multipleChan和minusChan，多出的兩個inputChan，就是用這個庫額外的開銷嘍。

原創不易，你的支持就是對我最大的鼓勵，歡迎給channelx點個star！:)

未完待續，channelx中還會陸續增長各類經常使用場景的channel實現，敬請期待……