快捷導(dǎo)航

Go語言協(xié)程處理數(shù)據(jù)有哪些問題

更新時間：2023年02月20日 09:59:44 作者：尋找09之夏

協(xié)程（coroutine）是Go語言中的輕量級線程實現(xiàn)，由Go運行時（runtime）管理。本文為大家詳細(xì)介紹了Go中的協(xié)程，協(xié)程不需要搶占式調(diào)度，可以有效提高線程的任務(wù)并發(fā)性，而避免多線程的缺點

前言

我們在開發(fā)后臺項目常常會遇到一個情況，功能模塊列表數(shù)據(jù)導(dǎo)出Excel功能，但列表中某個字段無法通過Sql聯(lián)表查詢，且一次性查詢再匹對也不方便；此時對列表數(shù)據(jù)循環(huán)，再一個個查詢結(jié)果加入列表，勢必需要很長的時間，我們該怎么才能提升下載速度呢？（這里采用Go開發(fā)服務(wù)端）

一、Goroutine

當(dāng)然第一個想到可能是采用協(xié)程處理循環(huán)里面要查詢的數(shù)據(jù)

type Card struct {
	Name    string  `json:"name"`
	Balance float64 `json:"balance"`
}
func main() {
	// 獲取卡列表數(shù)據(jù)
	list := getList()
	var data = make([]Card, 0, len(list))
	for _, val := range list {
		go func(card Card) {
			// 查詢業(yè)務(wù)，將值加入該記錄中
			var balance = getBalance()
			data = append(data, Card{
				Name:    card.Name,
				Balance: balance,
			})
		}(val)
	}
	log.Printf("數(shù)據(jù)：%+v", data)
}
// 獲取數(shù)據(jù)列表
func getList() []Card {
	var list = make([]Card, 0)
	for i := 0; i < 10000; i++ {
		list = append(list, Card{
			Name: "卡-" + strconv.Itoa(i+1),
		})
	}
	return list
}
// 獲取余額
func getBalance() float64 {
	time.Sleep(time.Millisecond * 100)
	return float64(rand.Int63n(1000))
}

運行上述代碼，結(jié)果： "數(shù)據(jù)：[]"，這是為什么呢？主要是協(xié)程處理業(yè)務(wù)需要時間，循環(huán)提前結(jié)束，所以才會出現(xiàn)這樣的結(jié)果，該怎么讓所有結(jié)果都處理結(jié)束才輸出結(jié)果呢？

二、sync.WaitGroup

此方法就是等待組進行多個任務(wù)的同步，等待組可以保證在并發(fā)環(huán)境中完成指定數(shù)量的任務(wù)

func main() {
	list := getList() // 獲取卡列表數(shù)據(jù)
	var data = make([]Card, 0, len(list))
	var wg sync.WaitGroup // 聲明一個等待組
	for _, val := range list {
		wg.Add(1) // 每一個任務(wù)開始時,將等待組增加1
		go func(card Card) {
			defer wg.Done() // 使用defer, 表示函數(shù)完成時將等待組值減1
			// 查詢業(yè)務(wù),休眠100微妙，將值加入該記錄中
			var balance = getBalance()
			data = append(data, Card{
				Name:    card.Name,
				Balance: balance,
			})
		}(val)
	}
	wg.Wait() // 等待所有任務(wù)完成
	log.Printf("數(shù)據(jù)：%+v", data)
}

運行結(jié)果會輸出所有數(shù)據(jù)，但細(xì)心的我們會發(fā)現(xiàn)，這個時候數(shù)據(jù)的順序是亂的，這個也符合業(yè)務(wù)需求，該怎么進一步改良呢？

三、數(shù)據(jù)排序

上面講到協(xié)程處理之后的額數(shù)據(jù)是無序的，這里我們知道數(shù)據(jù)跳數(shù)，直接初始化一個len和cap等于len(list)的空間，將之前append到data的數(shù)據(jù)改成通過下標(biāo)復(fù)制，這樣輸出的數(shù)據(jù)就是list的數(shù)據(jù)順序。

func main() {
	list := getList() // 獲取卡列表數(shù)據(jù)
	var data = make([]Card, len(list), len(list))
	var wg sync.WaitGroup // 聲明一個等待組
	for k, val := range list {
		wg.Add(1) // 每一個任務(wù)開始時,將等待組增加1
		go func(k int, card Card) {
			defer wg.Done() // 使用defer, 表示函數(shù)完成時將等待組值減1
			// 查詢業(yè)務(wù),休眠100微妙，將值加入該記錄中
			var balance = getBalance()
			data[k] = Card{
				Name:    card.Name,
				Balance: balance,
			}
		}(k, val)
	}
	wg.Wait() // 等待所有任務(wù)完成
	log.Printf("數(shù)據(jù)：%+v", data)
}

運行上述代碼，雖然可以獲取到想要的數(shù)據(jù)排序，但下次下載數(shù)據(jù)較多，開的協(xié)程過多，勢必導(dǎo)致資源開銷過大，帶來一系列問題，那怎么優(yōu)化限制協(xié)程個數(shù)呢？

四、限制協(xié)程數(shù)

大家都知道協(xié)程過多，自然消耗過多資源，可能導(dǎo)致其他問題；這里我們借助chan限制協(xié)程個數(shù)

// 限制100個協(xié)程
type pool struct {
	queue chan int
	wg    *sync.WaitGroup
}
func main() {
	list := getList() // 獲取卡列表數(shù)據(jù)
	var data = make([]Card, len(list), len(list))
	var gl = &pool{queue: make(chan int, 500), wg: &sync.WaitGroup{}} // 顯示協(xié)程數(shù)最大500個
	for k, val := range list {
		gl.queue <- 1 // 每一個任務(wù)開始時, chan輸入1個
		gl.wg.Add(1)  // 每一個任務(wù)開始時,將等待組增加1
		go func(k int, card Card) {
			defer func() {
				<-gl.queue   // 完成時chan取出1個
				gl.wg.Done() // 完成時將等待組值減1
			}()
			// 查詢業(yè)務(wù),休眠100微妙，將值加入該記錄中
			var balance = getBalance()
			data[k] = Card{
				Name:    card.Name,
				Balance: balance,
			}
		}(k, val)
	}
	gl.wg.Wait() // 等待所有任務(wù)完成
	log.Printf("數(shù)據(jù)：%+v", data)
}

通過使用chan，可以自己定義可協(xié)程最大數(shù)；現(xiàn)在看起來沒有什么問題，但如果協(xié)程獲取數(shù)據(jù)panic，會導(dǎo)致整個程序崩潰。

五、協(xié)程Panic處理

針對協(xié)程的panic（），我們需要接收，使用recover處理

func main() {
	list := getList() // 獲取卡列表數(shù)據(jù)
	var data = make([]Card, len(list), len(list))
	var gl = &pool{queue: make(chan int, 500), wg: &sync.WaitGroup{}} // 顯示協(xié)程數(shù)最大500個
	for k, val := range list {
		gl.queue <- 1 // 每一個任務(wù)開始時, chan輸入1個
		gl.wg.Add(1)  // 每一個任務(wù)開始時,將等待組增加1
		go func(k int, card Card) {
			// 解決協(xié)程panic，不至于程序崩潰
			defer func() {
				recover()
			}()
			defer func() {
				<-gl.queue   // 完成時chan取出1個
				gl.wg.Done() // 完成時將等待組值減1
			}()
			// 查詢業(yè)務(wù),休眠100微妙，將值加入該記錄中
			var balance = getBalance()
			data[k] = Card{
				Name:    card.Name,
				Balance: balance,
			}
		}(k, val)
	}
	gl.wg.Wait() // 等待所有任務(wù)完成
	log.Printf("數(shù)據(jù)：%+v", data)
}
// 獲取余額
func getBalance() float64 {
	panic("獲取余額panic")
	time.Sleep(time.Millisecond * 100)
	return float64(rand.Int63n(1000))
}

在協(xié)程中使用defer recover()；這樣協(xié)程拋出來的panic被接受，不會導(dǎo)致程序奔潰。