快捷導(dǎo)航

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

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

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

前言

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

一、Goroutine

當(dāng)然第一個(gè)想到可能是采用協(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))
}

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

二、sync.WaitGroup

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

func main() {
	list := getList() // 獲取卡列表數(shù)據(jù)
	var data = make([]Card, 0, len(list))
	var wg sync.WaitGroup // 聲明一個(gè)等待組
	for _, val := range list {
		wg.Add(1) // 每一個(gè)任務(wù)開始時(shí),將等待組增加1
		go func(card Card) {
			defer wg.Done() // 使用defer, 表示函數(shù)完成時(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)
}

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

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

上面講到協(xié)程處理之后的額數(shù)據(jù)是無序的，這里我們知道數(shù)據(jù)跳數(shù)，直接初始化一個(gè)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 // 聲明一個(gè)等待組
	for k, val := range list {
		wg.Add(1) // 每一個(gè)任務(wù)開始時(shí),將等待組增加1
		go func(k int, card Card) {
			defer wg.Done() // 使用defer, 表示函數(shù)完成時(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)
}

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

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

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

// 限制100個(gè)協(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個(gè)
	for k, val := range list {
		gl.queue <- 1 // 每一個(gè)任務(wù)開始時(shí), chan輸入1個(gè)
		gl.wg.Add(1)  // 每一個(gè)任務(wù)開始時(shí),將等待組增加1
		go func(k int, card Card) {
			defer func() {
				<-gl.queue   // 完成時(shí)chan取出1個(gè)
				gl.wg.Done() // 完成時(shí)將等待組值減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，會(huì)導(dǎo)致整個(gè)程序崩潰。

五、協(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個(gè)
	for k, val := range list {
		gl.queue <- 1 // 每一個(gè)任務(wù)開始時(shí), chan輸入1個(gè)
		gl.wg.Add(1)  // 每一個(gè)任務(wù)開始時(shí),將等待組增加1
		go func(k int, card Card) {
			// 解決協(xié)程panic，不至于程序崩潰
			defer func() {
				recover()
			}()
			defer func() {
				<-gl.queue   // 完成時(shí)chan取出1個(gè)
				gl.wg.Done() // 完成時(shí)將等待組值減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被接受，不會(huì)導(dǎo)致程序奔潰。