如何把golang的Channel玩出async和await的feel

引言

如何優雅的同步化異步代碼，一直以來都是各大編程語言致力於優化的點，記得最先是C# 5.0加入了async/await來簡化TPL的多線程模型，後來Javascript的Promise也吸收這一語法糖，在ES 6中也加入了async和await.

那麼，被你們一稱讚併發性能好、異步模型獨樹一幟的golang，可否也有async和await呢？

其實，這對於golang的CSM來講是一點也不難！

核心代碼以下：

done := make(chan struct{})
go func() {
	// do work asynchronously here
	//
	close(done)
}()
<-done
複製代碼

是否是很簡單呢？ go rountine負責async， channel的負責await, 簡直是完美！

但這個代碼看起來仍是有點醜，並且這個go func(){}尚未返回值，雖然說能夠經過閉包來接收返回值，但那個代碼就更難維護了。

Go Promise

代碼難看沒關係，只要Don't repeat yourself (DRY)，封裝一下不就行了？

type WorkFunc func() (interface{}, error) func NewPromise(workFunc WorkFunc) *Promise {
	promise := Promise{done: make(chan struct{})}
	go func() {
		defer close(promise.done)
		promise.res, promise.err = workFunc()
	}()
	return &promise
}

func (p *Promise) Done() (interface{}, error) {
	<-p.done
	return p.res, p.err
}
複製代碼

調用的代碼以下：

promise := NewPromise(func() (interface{}, error) {
	// do work asynchronously here
	//
	return res, err
})

// await
res, err := promise.Done()
複製代碼

是否是美觀了許多呢？

這個實現和Javascript的Promise的API是有很大差距，使用體驗上由於golang沒有泛型，也須要轉來轉去的，但爲了避免辜負Promise這個名字，怎麼能沒有then呢？

type SuccessHandler func(interface{}) (interface{}, error) type ErrorHandler func(error) interface{}

func (p *Promise) Then(successHandler SuccessHandler, errorHandler ErrorHandler) *Promise {
	newPromise := &Promise{done: make(chan struct{})}
	go func() {
		res, err := p.Done()
		defer close(newPromise.done)
		if err != nil {
			if errorHandler != nil {
				newPromise.res = errorHandler(err)
			} else {
				newPromise.err = err
			}
		} else {
			if successHandler != nil {
				newPromise.res, newPromise.err = successHandler(res)
			} else {
				newPromise.res = res
			}
		}
	}()

	return newPromise
}
複製代碼

有了then能夠chain起來，是否是找到些Promise的感受呢？

完整代碼請查看 promise.go

Actor

原本個人理解也就到些了，而後前段時間（說來也是一月有餘了），看了Go併發設計模式之 Active Object這篇文章後， 發現若是有一個常駐協程在異步的處理任務，並且是FIFO的，那麼這實際上是至關於一個無鎖的設計，能夠簡化對臨界資源的操做。

因而，我照着文章的思路，實現了下面的代碼：

// Creates a new actor
func NewActor(setActorOptionFuncs ...SetActorOptionFunc) *Actor {
	actor := &Actor{buffer: runtime.NumCPU(), quit: make(chan struct{}), wg: &sync.WaitGroup{}}
	for _, setOptionFunc := range setActorOptionFuncs {
		setOptionFunc(actor)
	}

	actor.queue = make(chan request, actor.buffer)

	actor.wg.Add(1)
	go actor.schedule()

	return actor
}

// The long live go routine to run.
func (actor *Actor) schedule() {
loop:
	for {
		select {
		case request := <-actor.queue:
			request.promise.res, request.promise.err = request.work()
			close(request.promise.done)
		case <-actor.quit:
			break loop
		}
	}
	actor.wg.Done()
}

// Do a work.
func (actor *Actor) Do(workFunc WorkFunc) *Promise {
	methodRequest := request{work: workFunc, promise: &Promise{
		done: make(chan struct{}),
	}}
	actor.queue <- methodRequest
	return methodRequest.promise
}

// Close actor
func (actor *Actor) Close() {
	close(actor.quit)
	actor.wg.Wait()
}
複製代碼

一個簡單的沒啥意義的純粹爲了demo的測試用例以下：

func TestActorAsQueue(t *testing.T) {
	actor := NewActor()
	defer actor.Close()

	i := 0
	workFunc := func() (interface{}, error) {
		time.Sleep(1 * time.Second)
		i++
		return i, nil
	}

	promise := actor.Do(workFunc)
	promise2 := actor.Do(workFunc)

	res2, _ := promise2.Done()
	res1, _ := promise.Done()

	if res1 != 1 {
		t.Fail()
	}

	if res2 != 2 {
		t.Fail()
	}
}
複製代碼

完整代碼請查看 actor.go

總結

每一個語言都有它的獨特之處，在個人理解中，玩轉golang的CSM模型，channel必定要用的6。

因而，我建立了Channelx這個repo, 包含了對channel經常使用場景的封裝，歡迎你們審閱，喜歡的就點個star。

此係列其它文章：

如何用Golang的channel實現消息的批量處理

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