快捷導(dǎo)航

Go實(shí)現(xiàn)后臺(tái)任務(wù)調(diào)度系統(tǒng)的實(shí)例代碼

更新時(shí)間：2023年06月12日 10:10:30 作者：堆棧future

平常我們?cè)陂_(kāi)發(fā)API的時(shí)候，前端傳遞過(guò)來(lái)的大批數(shù)據(jù)需要經(jīng)過(guò)后端處理，如果后端處理的速度快，前端響應(yīng)就快，反之則很慢，影響用戶(hù)體驗(yàn)，為了解決這一問(wèn)題，需要我們自己實(shí)現(xiàn)后臺(tái)任務(wù)調(diào)度系統(tǒng)，本文將介紹如何用Go語(yǔ)言實(shí)現(xiàn)后臺(tái)任務(wù)調(diào)度系統(tǒng),需要的朋友可以參考下

一、背景

平常我們?cè)陂_(kāi)發(fā)API的時(shí)候，前端傳遞過(guò)來(lái)的大批數(shù)據(jù)需要經(jīng)過(guò)后端處理，如果后端處理的速度快，前端響應(yīng)就快，反之則很慢，影響用戶(hù)體驗(yàn)。針對(duì)這種場(chǎng)景我們一般都是后臺(tái)異步處理，不需要前端等待所有的都執(zhí)行完才返回。為了解決這一問(wèn)題，需要我們自己實(shí)現(xiàn)后臺(tái)任務(wù)調(diào)度系統(tǒng)。

二、任務(wù)調(diào)度器實(shí)現(xiàn)

poll.go

package poller
import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"
)
type Poller struct {
	routineGroup *goroutineGroup // 并發(fā)控制
	workerNum    int // 記錄同時(shí)在運(yùn)行的最大goroutine數(shù)
	sync.Mutex
	ready  chan struct{} // 某個(gè)goroutine已經(jīng)準(zhǔn)備好了
	metric *metric // 統(tǒng)計(jì)當(dāng)前在運(yùn)行中的goroutine數(shù)量
}
func NewPoller(workerNum int) *Poller {
	return &Poller{
		routineGroup: newRoutineGroup(),
		workerNum:    workerNum,
		ready:        make(chan struct{}, 1),
		metric:       newMetric(),
	}
}
// 調(diào)度器
func (p *Poller) schedule() {
	p.Lock()
	defer p.Unlock()
	if int(p.metric.BusyWorkers()) >= p.workerNum {
		return
	}
	select {
	case p.ready <- struct{}{}: // 只要滿(mǎn)足當(dāng)前goroutine數(shù)量小于最大goroutine數(shù)量 那么就通知poll去調(diào)度goroutine執(zhí)行任務(wù)
	default:
	}
}
func (p *Poller) Poll(ctx context.Context) error {
	for {
		// step01
		p.schedule() // 調(diào)度
		select {
		case <-p.ready: // goroutine準(zhǔn)備好之后 這里就會(huì)有消息
		case <-ctx.Done():
			return nil
		}
	LOOP:
		for {
			select {
			case <-ctx.Done():
				break LOOP
			default:
				// step02
				task, err := p.fetch(ctx) // 獲取任務(wù)
				if err != nil {
					log.Println("fetch task error:", err.Error())
					break
				}
				fmt.Println(task)
				p.metric.IncBusyWorker() // 當(dāng)前正在運(yùn)行的goroutine+1
				// step03
				p.routineGroup.Run(func() { // 執(zhí)行任務(wù)
					if err := p.execute(ctx, task); err != nil {
						log.Println("execute task error:", err.Error())
					}
				})
				break LOOP
			}
		}
	}
}
func (p *Poller) fetch(ctx context.Context) (string, error) {
	time.Sleep(1000 * time.Millisecond)
	return "task", nil
}
func (p *Poller) execute(ctx context.Context, task string) error {
	defer func() {
		p.metric.DecBusyWorker() // 執(zhí)行完成之后 goroutine數(shù)量-1
		p.schedule() // 重新調(diào)度下一個(gè)goroutine去執(zhí)行任務(wù) 這一步是必須的
	}()
	return nil
}

metric.go

package poller
import "sync/atomic"
type metric struct {
	busyWorkers uint64
}
func newMetric() *metric {
	return &metric{}
}
func (m *metric) IncBusyWorker() uint64 {
	return atomic.AddUint64(&m.busyWorkers, 1)
}
func (m *metric) DecBusyWorker() uint64 {
	return atomic.AddUint64(&m.busyWorkers, ^uint64(0))
}
func (m *metric) BusyWorkers() uint64 {
	return atomic.LoadUint64(&m.busyWorkers)
}

goroutine_group.go

package poller
import "sync"
type goroutineGroup struct {
	waitGroup sync.WaitGroup
}
func newRoutineGroup() *goroutineGroup {
	return new(goroutineGroup)
}
func (g *goroutineGroup) Run(fn func()) {
	g.waitGroup.Add(1)
	go func() {
		defer g.waitGroup.Done()
		fn()
	}()
}
func (g *goroutineGroup) Wait() {
	g.waitGroup.Wait()
}

三、測(cè)試

package main
import (
	"context"
	"fmt"
	"ta/poller"
	"go.uber.org/goleak"
	"testing"
)
func TestMain(m *testing.M)  {
	fmt.Println("start")
	goleak.VerifyTestMain(m)
}
func TestPoller(t *testing.T) {
	producer := poller.NewPoller(5)
	producer.Poll(context.Background())
}

結(jié)果：

四、總結(jié)

大家用別的方式也可以實(shí)現(xiàn)，核心要點(diǎn)就是控制并發(fā)節(jié)奏，防止大量請(qǐng)求打到task service，在這里起到核心作用的就是schedule，它控制著整個(gè)任務(wù)系統(tǒng)的調(diào)度。同時(shí)還封裝了WaitGroup，這在大多數(shù)開(kāi)源代碼中都比較常見(jiàn)，大家可以去嘗試。另外就是test case一定得跟上，防止goroutine泄漏。

以上就是Go實(shí)現(xiàn)后臺(tái)任務(wù)調(diào)度系統(tǒng)的實(shí)例代碼的詳細(xì)內(nèi)容，更多關(guān)于Go后臺(tái)任務(wù)調(diào)度系統(tǒng)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: