goroutine并发实践(协程池+超时+错误快返回)

goroutinesync.WaitGroup{}

func testGoroutine() {
	wg := sync.WaitGroup{}
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			wg.Done()
			fmt.Println("hello world")
		}()
	}
	wg.Wait()
}
复制代码

goroutinerpchanggoroutinewg.Wait()

gorpc

hang

最简单的解法就是增加超时!

实际上超时也有很多解法

ctxcontext.WithTimeOut()select

select

func testWithGoroutineTimeOut() {
	var wg sync.WaitGroup
	done := make(chan struct{})
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
		}()
	}
	// wg.Wait()此时也要go出去,防止在wg.Wait()出堵住
	go func() {
		wg.Wait()
		close(done)
	}()
	select {
	// 正常结束完成
	case <-done:
	// 超时	
	case <-time.After(500 * time.Millisecond):
	}
}
复制代码

select

但是我们对于这个接口会有更高的要求。

goroutinegogoroutineforforgoroutine

goroutine

sync waitGroupchannel

package ezgopool

import "sync"

// goroutine pool
type GoroutinePool struct {
	c  chan struct{}
	wg *sync.WaitGroup
}

// 采用有缓冲channel实现,当channel满的时候阻塞
func NewGoroutinePool(maxSize int) *GoroutinePool {
	if maxSize <= 0 {
		panic("max size too small")
	}
	return &GoroutinePool{
		c:  make(chan struct{}, maxSize),
		wg: new(sync.WaitGroup),
	}
}

// add
func (g *GoroutinePool) Add(delta int) {
	g.wg.Add(delta)
	for i := 0; i < delta; i++ {
		g.c <- struct{}{}
	}

}

// done
func (g *GoroutinePool) Done() {
	<-g.c
	g.wg.Done()
}

// wait
func (g *GoroutinePool) Wait() {
	g.wg.Wait()
}

复制代码

golang

然后最后我们的超时+错误快返回+协程池模型就完成了~

func testGoroutineWithTimeOut() {
	 wg :=sync.WaitGroup{}
	done := make(chan struct{})
	// 新增阻塞chan
	errChan := make(chan error)

	pool.NewGoroutinePool(10)
	for i := 0; i < 10; i++ {
		pool.Add(1)
		go func() {
			pool.Done()
			if err!=nil{
				errChan<-errors.New("error")
			}
		}()
	}

	go func() {
		pool.Wait()
		close(done)
	}()

	select {
	// 错误快返回,适用于get接口
	case err := <-errChan:
		return nil, err
	case <-done:
	case <-time.After(500 * time.Millisecond):
	}
}

复制代码

谢谢