快捷導(dǎo)航

Golang?WorkerPool線程池并發(fā)模式示例詳解

更新時(shí)間：2022年08月17日 09:12:32 作者：龔國(guó)瑋

這篇文章主要為大家介紹了Golang?WorkerPool線程池并發(fā)模式示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

Worker Pools 線程池是一種并發(fā)模式。該模式中維護(hù)了固定數(shù)量的多個(gè)工作器，這些工作器等待著管理者分配可并發(fā)執(zhí)行的任務(wù)。該模式避免了短時(shí)間任務(wù)創(chuàng)建和銷(xiāo)毀線程的代價(jià)。

在 golang 中，我們使用 goroutine 和 channel 來(lái)構(gòu)建這種模式。工作器 worker 由一個(gè) goroutine 定義，該 goroutine 通過(guò) channel 獲取數(shù)據(jù)。

處理CVS文件記錄

接下來(lái)讓我們通過(guò)一個(gè)例子，來(lái)進(jìn)一步理解該模式。假設(shè)您需要處理來(lái)自 CVS 文件的記錄數(shù)據(jù)，我們需要將該文件中的經(jīng)緯度保存到數(shù)據(jù)庫(kù)中。代碼如下。

package main
import (
	"encoding/csv"
	"fmt"
	"os"
	"time"
)
type city struct {
	name     string
	location string
}
func createCity(record city) {
	time.Sleep(10 * time.Millisecond)
}
func main() {
	startTime := time.Now()
	csvFile, err := os.Open("cities.csv")
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("Successfully Opened CSV file")
	defer csvFile.Close()
	csvLines, err := csv.NewReader(csvFile).ReadAll()
	if err != nil {
		fmt.Println(err)
	}
	counter := 0
	for _, line := range csvLines {
		counter++
		createCity(city{
			name:     line[0],
			location: line[1],
		})
	}
	fmt.Println("records saved:", counter)
	fmt.Println("total time:", time.Since(startTime))
}

?? 獲取測(cè)試數(shù)據(jù)

cities.csv

輸出：

正如我們所看到的，保存 CSV 中所有記錄需要 55 秒，這是很長(zhǎng)的時(shí)間，可能會(huì)導(dǎo)致很多性能問(wèn)題。用戶如果想要上傳 CSV 文件，那體驗(yàn)感一定很差。

如何解決這個(gè)問(wèn)題？那我們就使用線程池的方法試試看。

線程池耗時(shí)差異

在如下示例中，我們將解決相同的需求，但通過(guò)線程池，耗時(shí)方面，我們能夠看到巨大的差異。來(lái)吧！

代碼如下

package main
import (
	"encoding/csv"
	"fmt"
	"os"
	"time"
)
type city struct {
	name     string
	location string
}
func createCity(record city) {
	time.Sleep(10 * time.Millisecond)
}
func readData(cityChn chan []city) {
	var cities []city
	csvFile, err := os.Open("cities.csv")
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("Successfully Opened CSV file")
	defer csvFile.Close()
	csvLines, err := csv.NewReader(csvFile).ReadAll()
	if err != nil {
		fmt.Println(err)
	}
	for _, line := range csvLines {
		cities = append(cities, city{
			name:     line[0],
			location: line[1],
		})
	}
	cityChn <- cities
}
func worker(cityChn chan city) {
	for val := range cityChn {
		createCity(val)
	}
}
func main() {
	startTime := time.Now()
	cities := make(chan []city)
	go readData(cities)
	const workers = 5
	jobs := make(chan city, 1000)
	for w := 1; w <= workers; w++ {
		go worker(jobs)
	}
	counter := 0
	for _, val := range <-cities {
		counter++
		jobs <- val
	}
	fmt.Println("records saved:", counter)
	fmt.Println("total time:", time.Since(startTime))
}

輸出：

你看到很大的不同了嗎？現(xiàn)在同樣的過(guò)程只需要 8 秒。正如您所見(jiàn)，當(dāng)我們需要處理大量數(shù)據(jù)時(shí)，線程池非常有用。

使用線程池，我們必須定義一個(gè)函數(shù)，在示例中該函數(shù)為 worker，該函數(shù)用于定義工作進(jìn)程，您可以看到它接收一個(gè) Channel 通道來(lái)處理數(shù)據(jù)。另外，我們必須在數(shù)據(jù)傳遞到通道之前啟動(dòng) goroutines 協(xié)程，當(dāng) Channel 通道獲取到值時(shí)，goroutines 工作者開(kāi)始處理它們。

?? 現(xiàn)在您知道如何實(shí)現(xiàn)線程池了！

以上就是Golang WorkerPool線程池并發(fā)模式示例詳解的詳細(xì)內(nèi)容，更多關(guān)于Golang WorkerPool線程池并發(fā)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: