golang開發:go并發的建議(完)

上次說了一下Go語言布道師 Dave Cheney對Go并發的建議，個人覺得最重要的一條，這次主要想說一下這個，
8.3. Never start a goroutine without knowning when it will stop(永遠不要在不知道何時停止的情況下啟動 goroutine)

我們的需求

我這邊當時有個需求是這樣的，我們有個考試系統的，每次學員答完試卷去檢查一下這次交卷是否是這次考試的最后一份試卷，如果是最后一份試卷的話，需要計算這次考試的總成績，生成考試的學習報告，當然了，如果不是最后一份試卷的話啥也不干，
生成試卷和報告是必須要生成的，不能出現考完試了沒有總成績和總報告，
接到這個需求的時候，我首先想到的是使用golang的goroutine去異步算出成績生成報告，然后寫代碼就是這樣的，

go createReport()

這不剛好是8.3 永遠不要這樣寫的建議么？
然后覺得應該寫一個管理goroutine異步執行任務的類別庫，創建執行銷毀都由這個管理工具去執行，準備寫的時候發現B站的代碼里有一個這樣的類別庫，異步執行的類別庫，

B站的類別庫

B站代碼里面異步任務是這個檔案
openbilibili-go-common-master/library/sync/pipeline/fanout/fanout.go

var (
	// ErrFull chan full.
	ErrFull   = errors.New("fanout: chan full")
	stats     = prom.BusinessInfoCount
	traceTags = []trace.Tag{
		trace.Tag{Key: trace.TagSpanKind, Value: "background"},
		trace.Tag{Key: trace.TagComponent, Value: "sync/pipeline/fanout"},
	}
)

type options struct {
	worker int
	buffer int
}

// Option fanout option
type Option func(*options)

// Worker specifies the worker of fanout
func Worker(n int) Option {
	if n <= 0 {
		panic("fanout: worker should > 0")
	}
	return func(o *options) {
		o.worker = n
	}
}

// Buffer specifies the buffer of fanout
func Buffer(n int) Option {
	if n <= 0 {
		panic("fanout: buffer should > 0")
	}
	return func(o *options) {
		o.buffer = n
	}
}

type item struct {
	f   func(c context.Context)
	ctx context.Context
}

// Fanout async consume data from chan.
type Fanout struct {
	name    string
	ch      chan item
	options *options
	waiter  sync.WaitGroup

	ctx    context.Context
	cancel func()
}

// New new a fanout struct.
func New(name string, opts ...Option) *Fanout {
	if name == "" {
		name = "fanout"
	}
	o := &options{
		worker: 1,
		buffer: 1024,
	}
	for _, op := range opts {
		op(o)
	}
	c := &Fanout{
		ch:      make(chan item, o.buffer),
		name:    name,
		options: o,
	}
	c.ctx, c.cancel = context.WithCancel(context.Background())
	c.waiter.Add(o.worker)
	for i := 0; i < o.worker; i++ {
		go c.proc()
	}
	return c
}

func (c *Fanout) proc() {
	defer c.waiter.Done()
	for {
		select {
		case t := <-c.ch:
			wrapFunc(t.f)(t.ctx)
			stats.State(c.name+"_channel", int64(len(c.ch)))
		case <-c.ctx.Done():
			return
		}
	}
}

func wrapFunc(f func(c context.Context)) (res func(context.Context)) {
	res = func(ctx context.Context) {
		defer func() {
			if r := recover(); r != nil {
				buf := make([]byte, 64*1024)
				buf = buf[:runtime.Stack(buf, false)]
				log.Error("panic in fanout proc, err: %s, stack: %s", r, buf)
			}
		}()
		f(ctx)
		if tr, ok := trace.FromContext(ctx); ok {
			tr.Finish(nil)
		}
	}
	return
}

// Do save a callback func.
func (c *Fanout) Do(ctx context.Context, f func(ctx context.Context)) (err error) {
	if f == nil || c.ctx.Err() != nil {
		return c.ctx.Err()
	}
	nakeCtx := metadata.WithContext(ctx)
	if tr, ok := trace.FromContext(ctx); ok {
		tr = tr.Fork("", "Fanout:Do").SetTag(traceTags...)
		nakeCtx = trace.NewContext(nakeCtx, tr)
	}
	select {
	case c.ch <- item{f: f, ctx: nakeCtx}:
	default:
		err = ErrFull
	}
	stats.State(c.name+"_channel", int64(len(c.ch)))
	return
}

// Close close fanout
func (c *Fanout) Close() error {
	if err := c.ctx.Err(); err != nil {
		return err
	}
	c.cancel()
	c.waiter.Wait()
	return nil
}

使用方法
	ca := New("cache", Worker(100), Buffer(1024))
	var run bool
	ca.Do(context.Background(), func(c context.Context) {
		run = true
	})

主要分析一下這個類別庫，以后自己寫或者使用的時候就能得心應手了，而且這個類別庫也算是創建goroutine，通過channel通信的經典寫法吧

1.New方法呼叫的時候，會創建buffer個ch channel，worker個goroutine.由于ch是空的，worker個goroutine會阻塞住，一直等待有程式往ch里面寫入資料
2.Do函式一但被呼叫，會傳入異步任務的func，func就會寫入到ch里面了，goroutine就可以從ch里面讀取到資料，并且執行這個資料里面的func
踐行了這個原則
不要通過共享記憶體來通信,要通過通信來共享記憶體

有個需要注意的點，就Do函式在執行代碼是這樣的

代碼里面可以看到在c.ch 寫入資料的時候，如果超過c.ch的長度（測驗代碼里面是1024）就報錯回傳了，這樣就不能保證每個異步任務都能穩定執行了，這樣的結果就是，如果程式處理慢或者異步任務數量比較多的話（超過1024），異步任務就無法完成，當然了，我們也可以修改代碼改成等待ch的里面資料被goroutine處理的小于1024了，也會執行，這樣就變成一個不可控的程式了，如果有3000個異步任務沒人知道執行完成需要多長時間，然后我們程式如果重啟的話，是等待它完成重啟還是強制重啟，等待完成不知道需要等待多長時間，強制重啟就無法保證任務能夠全部完成，

最終方案

為了一定能夠在任何例外情況算出分數和生成報告，最后使用訊息佇列做了這件事，發送完成答卷的訊息，接收到完成答卷的訊息之后算出分數生成報告，做完之后雖然保證了可靠性，但是覺得自己發訊息自己收訊息確實也很別扭，
不知道其他童鞋有沒有更好的更合理的方案，

標籤：Go

上一篇：垃圾回收

下一篇：golang開發:go并發的建議(完)