快捷導(dǎo)航

Golang實(shí)現(xiàn)帶優(yōu)先級(jí)的select

更新時(shí)間：2023年04月02日 09:42:56 作者：jxwu

這篇文章主要為大家詳細(xì)介紹了如何在Golang中實(shí)現(xiàn)帶優(yōu)先級(jí)的select，文中的示例代碼講解詳細(xì)，對(duì)我們學(xué)習(xí)Golang有一定的幫助，需要的可以參考一下

背景

在 Golang 里面，我們經(jīng)常使用 channel 進(jìn)行協(xié)程之間的通信。這里有一個(gè)經(jīng)典的場(chǎng)景，也就是生產(chǎn)者消費(fèi)者模式，生產(chǎn)者協(xié)程不斷地往 Channel 里面塞元素，而消費(fèi)者協(xié)程不斷地消費(fèi)這些元素。

寫成代碼就是如下：

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()
}

// 生產(chǎn)者
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)
	}
}

// 消費(fèi)者
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) // 消費(fèi)元素
	}
}

生產(chǎn)者不斷產(chǎn)生元素，消費(fèi)者消費(fèi)元素。生產(chǎn)者不會(huì)等待消費(fèi)者消費(fèi)完畢（不然可能影響其他任務(wù)），如果 channel 已經(jīng)滿了，也就是說明消費(fèi)者消費(fèi)不過來，生產(chǎn)者就會(huì)丟棄這個(gè)任務(wù)。

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

// 消費(fèi)者
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) // 消費(fèi)元素:
		case <-t.C:
			clear() // 清理
		}
	}
}

運(yùn)行程序到第15秒的時(shí)候，生產(chǎn)者發(fā)現(xiàn) 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

解決方案

既然清理任務(wù)的優(yōu)先級(jí)并不高，那么它就不應(yīng)該阻塞消費(fèi)元素流程，而是應(yīng)該在空閑時(shí)才去執(zhí)行。由于 Golang 里面，如果 select 兩個(gè) case 都同時(shí)滿足，會(huì)隨機(jī)選一個(gè)執(zhí)行，因此第一想到的可能會(huì)使用如下代碼實(shí)現(xiàn)優(yōu)先級(jí)case：

// 消費(fèi)者
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) // 消費(fèi)元素
			continue   // 可能還有元素，不走清理邏輯
		default:
		}

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

如果運(yùn)行這個(gè)程序，可以發(fā)現(xiàn)它能夠滿足優(yōu)先級(jí)的需求，先消費(fèi)元素，空閑時(shí)再執(zhí)行清理任務(wù)。

然而，在沒有元素可以消費(fèi)，也沒有清理任務(wù)可以執(zhí)行的時(shí)候，這里的for將會(huì)不斷地循環(huán)，浪費(fèi)CPU資源。

其實(shí)，可以使用下面的方法實(shí)現(xiàn)優(yōu)先級(jí)case，它能夠在沒有元素就緒的時(shí)候阻塞在 select，而不是不斷循環(huán)：

// 消費(fèi)者
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) // 消費(fèi)元素
		case <-t.C:
		priority:
			for { // 清理前先把元素消費(fèi)完
				select {
				case i := <-ch:
					consume(i) // 消費(fèi)元素
				default:
					break priority // 注：這里會(huì)跳過這個(gè)循環(huán)，而不是再次執(zhí)行
				}
			}
			clear() // 清理
		}
	}
}

這里的關(guān)鍵是在觸發(fā)清理case的時(shí)候，先去把channel里面的元素消費(fèi)完，再進(jìn)行清理，從而保證能夠留下足夠的channel緩沖區(qū)給生產(chǎn)者放置生產(chǎn)的元素。

一個(gè)封裝

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

// 優(yōu)先級(jí)select ch1 的任務(wù)先執(zhí)行完畢后才會(huì)執(zhí)行 ch2 里面的任務(wù)
func PrioritySelect[T1, T2 any](ch1 <-chan T1, f1 func(T1), ch2 <-chan T2, f2 func(T2)) {
	for {
		select {
		case a := <-ch1:
			f1(a)
		case b := <-ch2:
		priority:
			for {
				select {
				case a := <-ch1:
					f1(a)
				default:
					break priority
				}
			}
			f2(b)
		}
	}
}

這樣，我們的消費(fèi)者代碼就可以簡化為：

// 消費(fèi)者
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)
}

到此這篇關(guān)于Golang實(shí)現(xiàn)帶優(yōu)先級(jí)的select的文章就介紹到這了,更多相關(guān)Golang帶優(yōu)先級(jí)select內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: