Golang怎么實現帶優先級的select

這篇“Golang怎么實現帶優先級的select”文章的知識點大部分人都不太理解，所以小編給大家總結了以下內容，內容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Golang怎么實現帶優先級的select”文章吧。

背景

在 Golang 里面，我們經常使用 channel 進行協程之間的通信。這里有一個經典的場景，也就是生產者消費者模式，生產者協程不斷地往 Channel 里面塞元素，而消費者協程不斷地消費這些元素。

寫成代碼就是如下：

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
	ch := make(chan int, 10)
	var wg sync.WaitGroup
	wg.Add(2)
	go producer(ch, &wg)
	go consumer(ch, &wg)
	wg.Wait()
}

// 生產者
func producer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	i := 0
	for {
		select {
		case ch <- i:
		default:
			// 丟棄
			log.Println("discard")
		}
		i++
		time.Sleep(time.Second)
	}
}

// 消費者
func consumer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	consume := func(i int) {
		fmt.Println(i)
		time.Sleep(time.Millisecond * 700)
	}
	for {
		i := <-ch
		consume(i) // 消費元素
	}
}

生產者不斷產生元素，消費者消費元素。生產者不會等待消費者消費完畢（不然可能影響其他任務），如果 channel 已經滿了，也就是說明消費者消費不過來，生產者就會丟棄這個任務。

生產者平均一秒生成1個，消費者0.7秒消費一個。正常情況下消費者是消費得過來的，然而很多時候消費者協程還需要做一些定時任務，比如一些定時清理工作。假如這個清理工作每2秒觸發一次，清理時間一般需要1.5秒，也就是如果每次都做每一秒有0.75秒會被清理工作占有了，但是它不是一定要非常及時的，可以等空閑時再進行。 如下代碼：

// 消費者
func consumer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	t := time.NewTicker(time.Second * 2)
	consume := func(i int) {
		fmt.Println(i)
		time.Sleep(time.Millisecond * 700)
	}
	clear := func() {
		fmt.Println("clear")
		time.Sleep(time.Millisecond * 1500)
	}
	for {
		select {
		case i := <-ch:
			consume(i) // 消費元素:
		case <-t.C:
			clear() // 清理
		}
	}
}

運行程序到第15秒的時候，生產者發現 channel滿了，于是開始丟包：

0
1
clear
2
3
4
5
6
clear
7
clear
8
clear
9
clear
clear
10
clear
11
12
13
14
clear
15
clear
clear
discard
16
clear
discard
discard

解決方案

既然清理任務的優先級并不高，那么它就不應該阻塞消費元素流程，而是應該在空閑時才去執行。由于 Golang 里面，如果 select 兩個 case 都同時滿足，會隨機選一個執行，因此第一想到的可能會使用如下代碼實現優先級case：

// 消費者
func consumer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	t := time.NewTicker(time.Second * 2)
	consume := func(i int) {
		fmt.Println(i)
		time.Sleep(time.Millisecond * 700)
	}
	clear := func() {
		fmt.Println("clear")
		time.Sleep(time.Millisecond * 1500)
	}
	for {
		select {
		case i := <-ch:
			consume(i) // 消費元素
			continue   // 可能還有元素，不走清理邏輯
		default:
		}

		// 沒有元素才走清理邏輯
		select {
		case <-t.C:
			clear() // 清理
		default:
		}
	}
}

如果運行這個程序，可以發現它能夠滿足優先級的需求，先消費元素，空閑時再執行清理任務。

然而，在沒有元素可以消費，也沒有清理任務可以執行的時候，這里的for將會不斷地循環，浪費CPU資源。

其實，可以使用下面的方法實現優先級case，它能夠在沒有元素就緒的時候阻塞在 select，而不是不斷循環：

// 消費者
func consumer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	t := time.NewTicker(time.Second * 2)
	consume := func(i int) {
		fmt.Println(i)
		time.Sleep(time.Millisecond * 700)
	}
	clear := func() {
		fmt.Println("clear")
		time.Sleep(time.Millisecond * 1500)
	}
	for {
		select {
		case i := <-ch:
			consume(i) // 消費元素
		case <-t.C:
		priority:
			for { // 清理前先把元素消費完
				select {
				case i := <-ch:
					consume(i) // 消費元素
				default:
					break priority // 注：這里會跳過這個循環，而不是再次執行
				}
			}
			clear() // 清理
		}
	}
}

這里的關鍵是在觸發清理case的時候，先去把channel里面的元素消費完，再進行清理，從而保證能夠留下足夠的channel緩沖區給生產者放置生產的元素。

一個封裝

上面那段優先級case代碼其實挺常用的，但是幾乎都是模板代碼，特別是需要在兩個地方寫consume(i)，因此我們可以封裝一下這段代碼，方便使用，減少出錯：

// 優先級select ch2 的任務先執行完畢后才會執行 ch3 里面的任務
func PrioritySelect[T1, T2 any](ch2 <-chan T1, f1 func(T1), ch3 <-chan T2, f2 func(T2)) {
	for {
		select {
		case a := <-ch2:
			f1(a)
		case b := <-ch3:
		priority:
			for {
				select {
				case a := <-ch2:
					f1(a)
				default:
					break priority
				}
			}
			f2(b)
		}
	}
}

這樣，我們的消費者代碼就可以簡化為：

// 消費者
func consumer(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	t := time.NewTicker(time.Second * 2)
	consume := func(i int) {
		fmt.Println(i)
		time.Sleep(time.Millisecond * 700)
	}
	clear := func(time.Time) {
		fmt.Println("clear")
		time.Sleep(time.Millisecond * 1500)
	}
	PrioritySelect(ch, consume, t.C, clear)
}

以上就是關于“Golang怎么實現帶優先級的select”這篇文章的內容，相信大家都有了一定的了解，希望小編分享的內容對大家有幫助，若想了解更多相關的知識內容，請關注億速云行業資訊頻道。