GO語言怎么實現協程池管理

本篇內容介紹了“GO語言怎么實現協程池管理”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

使用channel實現協程池

通過 Channel 實現 Goroutine Pool,缺點是會造成協程的頻繁開辟和注銷，但好在簡單靈活通用。

package main

import (
	"fmt"
	"io/ioutil"
	"net/http"
	"sync"
)

// Pool goroutine Pool
type Pool struct {
	queue chan int
	wg    *sync.WaitGroup
}

// New 新建一個協程池
func New(size int) *Pool {
	if size <= 0 {
		size = 1
	}
	return &Pool{
		queue: make(chan int, size),
		wg:    &sync.WaitGroup{},
	}
}

// Add 新增一個執行
func (p *Pool) Add(delta int) {
	// delta為正數就添加
	for i := 0; i < delta; i++ {
		p.queue <- 1
	}
	// delta為負數就減少
	for i := 0; i > delta; i-- {
		<-p.queue
	}
	p.wg.Add(delta)
}

// Done 執行完成減一
func (p *Pool) Done() {
	<-p.queue
	p.wg.Done()
}

func (p *Pool) Wait() {
	p.wg.Wait()
}

func main() {
	// 這里限制100個并發
	pool := New(100) // sync.WaitGroup{}

	//假設需要發送1000萬個http請求，然后我并發100個協程取完成這件事
	for i := 0; i < 10000000; i++ {
		pool.Add(1) //發現已存在100個人正在發了，那么就會卡住，直到有人完成了宣布自己退出協程了
		go func(i int) {
			resp, err := http.Get("https://www.baidu.com")
			if err != nil {
				fmt.Println(i, err)
			} else {
				defer resp.Body.Close()
				result, _ := ioutil.ReadAll(resp.Body)
				fmt.Println(i, string(result))
			}
			pool.Done()
		}(i)
	}
	pool.Wait()
}

消費者模式實現協程池

頻繁對協程開辟與剔除，如果對性能有著很高的要求，建議優化成固定數目的協程取 channel 里面取數據進行消費，這樣可以避免協程的創建與注銷。

package main

import (
	"fmt"
	"strconv"
	"sync"
)

// 任務對象
type task struct {
	Production
	Consumer
}

// 設置消費者數目，也就是work pool大小
func (t *task) setConsumerPoolSize(poolSize int) {
	t.Production.Jobs = make(chan *Job, poolSize*10)
	t.Consumer.WorkPoolNum = poolSize
}

// 任務數據對象
type Job struct {
	Data string
}

func NewTask(handler func(jobs chan *Job) (b bool)) (t *task) {
	t = &task{
		Production: Production{Jobs: make(chan *Job, 100)},
		Consumer:   Consumer{WorkPoolNum: 10, Handler: handler},
	}
	return
}

type Production struct {
	Jobs chan *Job
}

func (c Production) AddData(data *Job) {
	c.Jobs <- data
}

type Consumer struct {
	WorkPoolNum int
	Handler     func(chan *Job) (b bool)
	Wg          sync.WaitGroup
}

// 異步開啟多個work去處理任務，但是所有work執行完畢才會退出程序
func (c *Consumer) disposeData(data chan *Job) {
	for i := 0; i <= c.WorkPoolNum; i++ {
		c.Wg.Add(1)
		go func() {
			defer func() {
				c.Wg.Done()
			}()
			c.Handler(data)
		}()
	}
	c.Wg.Wait()
}

func main() {
	// 實現一個用于處理數據的閉包，實現業務代碼
	consumerHandler := func(jobs chan *Job) (b bool) {
		for jobs := range jobs {
			fmt.Println(jobs)
		}
		return
	}

	// new一個任務處理對象
	t := NewTask(consumerHandler)
	t.setConsumerPoolSize(500) // 500個協程同時消費

	// 根據自己的業務去生成數據通過AddData方法添加數據到生成channel，這里是100萬條數據
	go func() {
		for i := 0; i < 1000000; i++ {
			job := new(Job)
			iStr := strconv.Itoa(i)
			job.Data = "定義任務數據格式" + iStr
			t.AddData(job)
		}
	}()

	// 消費者消費數據
	t.Consumer.disposeData(t.Production.Jobs)
}

“GO語言怎么實現協程池管理”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！