Go http請求排隊處理實戰示例

Posted on by

一、http請求的順序處理方式

在高並發場景下，為瞭降低系統壓力，都會使用一種讓請求排隊處理的機制。本文就介紹在Go中是如何實現的。

首先，我們看下正常的請求處理邏輯。 客戶端發送請求，web server接收請求，然後就是處理請求，最後響應給客戶端這樣一個順序的邏輯。如下圖所示：

代碼實現如下：

package main
import (
	"fmt"
	"net/http"
)
func main() {
	myHandler := MyHandler{}
	http.Handle("/", &myHandler)
	http.ListenAndServe(":8080", nil)
}
type MyHandler struct {
}
func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	w.Write([]byte("Hello Go"))
}

在瀏覽器中輸入 http://localhost:8080/，就能在頁面上顯示出“Hello Go”的頁面來。

通常情況下，大傢在開發web系統的時候，一般都是這麼處理請求。接下來我們看在高並發下如何實現讓請求進行排隊處理。

二、http請求的異步處理方式–排隊處理

讓http請求進入到隊列，我們也稱為異步處理方式。其基本思想就是將接收到的請求的上下文（即request和response）以及處理邏輯包裝成一個工作單元，然後將其放到隊列，然後該工作單元等待消費的工作線程處理該job，處理完成後再返回給客戶端。 流程如下圖：

該實現中會有三個關鍵的元素：工作執行單元、隊列、消費者。下面我們逐一看下各自的職責及實現。

工作單元

該工作單元主要是封裝請求的上下文信息（request和response）、請求的處理邏輯以及該工作單元是否被執行完成的狀態。

請求的處理邏輯實際上就是原來在順序處理流程中的具體函數，如果是mvc模式的話就是controller裡的一個具體的action。

在Go中實現通信的方式一般是使用通道。所以，在工作單元中有一個通道，當該工作單元執行完具體的處理邏輯後，就往該通道中寫入一個消息，以通知主協程該次請求已完成，可以返回給客戶端瞭。

所以，一個http請求的處理邏輯看起來就像是下面這樣：

func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  將w和r包裝成工作單元job
  將job入隊
  等待job執行完成
  本次請求處理完畢
}

下面我們看下工作單元的具體實現，這裡我們將其定義為一個Job結構體：

type Job struct {
    DoneChan  chan struct{}
    handleJob func(j FlowJob) error //具體的處理邏輯
}

Job結構體中有一個handleJob，其類型是一個函數，即處理請求的邏輯部分。DoneChan通道用來讓該單元進行阻塞等待，並當handleJob執行完畢後發送消息通知的。

下面我們再看看該Job的相關行為：

// 消費者從隊列中取出該job時 執行具體的處理邏輯
func (job *Job) Execute() error {
    fmt.Println("job start to execute ")
    return job.handleJob(job)
}
// 執行完Execute後，調用該函數以通知主線程中等待的job
func (job *Job) Done() {
    job.DoneChan <- struct{}{}
    close(job.DoneChan)
}
// 工作單元等待自己被消費
func (job *Job) WaitDone() {
    select {
    case <-job.DoneChan:
	return
    }
}

隊列

隊列主要是用來存儲工作單元的。是處理請求的主協程和消費協程之間的紐帶。隊列具有列表、容量、當前元素個數等關鍵元素組成。如下：

type JobQueue struct {
    mu         sync.Mutex
    noticeChan chan struct{}
    queue      *list.List
    size       int
    capacity   int
}

其行為主要有入隊、出隊、移除等操作。定義如下：

// 初始化隊列
func NewJobQueue(cap int) *JobQueue {
    return &JobQueue{
	capacity: cap,
	queue:    list.New(),
	noticeChan: make(chan struct{}, 1),
    }
}
// 工作單元入隊
func (q *JobQueue) PushJob(job *Job) {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.size++
    if q.size > q.capacity {
	q.RemoveLeastJob()
    }
    q.queue.PushBack(job)
    q.noticeChan <- struct{}{}
}
// 工作單元出隊
func (q *JobQueue) PopJob() *Job {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.size == 0 {
		return nil
	}
	q.size--
	return q.queue.Remove(q.queue.Front()).(*Job)
}
// 移除隊列中的最後一個元素。
// 一般在容量滿時，有新job加入時，會移除等待最久的一個job
func (q *JobQueue) RemoveLeastJob() {
	if q.queue.Len() != 0 {
		back := q.queue.Back()
		abandonJob := back.Value.(*Job)
		abandonJob.Done()
		q.queue.Remove(back)
	}
}
// 消費線程監聽隊列的該通道，查看是否有新的job需要消費
func (q *JobQueue) waitJob() <-chan struct{} {
    return q.noticeChan
}

這裡我們主要解釋一下入隊的操作流程：

  • 1 首先是隊列的元素個數size++
  • 2 判斷size是否超過最大容量capacity
  • 3 若超過最大容量，則將隊列中最後一個元素移除。因為該元素等待時間最長，認為是超時的情況。
  • 4 將新接收的工作單元放入到隊尾。
  • 5 往noticeChan通道中寫入一個消息，以便通知消費協程處理Job。

由以上可知，noticeChan是隊列和消費者協程之間的紐帶。下面我們來看看消費者的實現。

消費者協程

消費者協程的職責是監聽隊列，並從隊列中獲取工作單元，執行工作單元的具體處理邏輯。在實際應用中，可以根據系統的承載能力啟用多個消費協程。在本文中，為瞭方便講解，我們隻啟用一個消費協程。

我們定義一個WorkerManager結構體，負責管理具體的消費協程。該WorkerManager有一個屬性是工作隊列，所有啟動的消費協程都需要從該工作隊列中獲取工作單元。代碼實現如下：

type WorkerManager struct {
    jobQueue *JobQueue
}
func NewWorkerManager(jobQueue *JobQueue) *WorkerManager {
    return &WorkerManager{
	jobQueue: jobQueue,
    }
}
func (m *WorkerManager) createWorker() error {
    go func() {
	fmt.Println("start the worker success")
	var job FlowJob
	for {
            select {
                case <-m.jobQueue.waitJob():
		fmt.Println("get a job from job queue")
                job = m.jobQueue.PopJob()
		fmt.Println("start to execute job")
		job.Execute()
                fmt.Print("execute job done")
		job.Done()
            }
	}
    }()
    return nil
}

在代碼中我們可以看到，createWorker中的邏輯實際是一個for循環，然後通過select監聽隊列的noticeChan通道，當獲取到工作單元時，就執行工作單元中的handleJob方法。執行完後，通過job.Done()方法通知在主協程中還等待的job。這樣整個流程就形成瞭閉環。

完整代碼

我們現在看下整體的處理流程，如下圖：

現在我們寫一個測試demo。在這裡我們定義瞭一個全局的flowControl結構體，以作為隊列和工作協程的管理。代碼如下：

package main
import (
    "container/list"
    "fmt"
    "net/http"
    "sync"
)
func main() {
    flowControl := NewFlowControl()
    myHandler := MyHandler{
	flowControl: flowControl,
    }
    http.Handle("/", &myHandler)
    http.ListenAndServe(":8080", nil)
}
type MyHandler struct {
    flowControl *FlowControl
}
func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	fmt.Println("recieve http request")
	job := &Job{
            DoneChan: make(chan struct{}, 1),
            handleJob: func(job *Job) error {
		w.Header().Set("Content-Type", "application/json")
		w.Write([]byte("Hello World"))
		return nil
            },
	}
	h.flowControl.CommitJob(job)
	fmt.Println("commit job to job queue success")
	job.WaitDone()
}
type FlowControl struct {
    jobQueue *JobQueue
    wm       *WorkerManager
}
func NewFlowControl() *FlowControl {
    jobQueue := NewJobQueue(10)
    fmt.Println("init job queue success")
    m := NewWorkerManager(jobQueue)
    m.createWorker()
    fmt.Println("init worker success")
    control := &FlowControl{
	jobQueue: jobQueue,
	wm:       m,
    }
    fmt.Println("init flowcontrol success")
    return control
}
func (c *FlowControl) CommitJob(job *Job) {
    c.jobQueue.PushJob(job)
    fmt.Println("commit job success")
}

之前有一篇文章是優先級隊列，實際上就是該隊列的高級實現版本，可以將不同的請求按優先級分配到不同的隊列中。有興趣的同學可參考：Go實戰 單隊列到優先級隊列的實現

總結

通過將請求的上下文信息封裝到一個工作單元中，並將其放入到隊列中，然後通過消息通道的方式阻塞等待消費者執行完畢。同時在隊列中通過設置隊列的容量以解決請求過多而給系統造成壓力的問題。

以上就是Go http請求排隊處理實戰的詳細內容，更多關於Go http請求排隊的資料請關註WalkonNet其它相關文章！

推薦閱讀：