Go?http請求排隊處理實戰(zhàn)示例
一、http請求的順序處理方式
在高并發(fā)場景下，為了降低系統(tǒng)壓力，都會使用一種讓請求排隊處理的機制。本文就介紹在Go中是如何實現(xiàn)的。
首先，我們看下正常的請求處理邏輯。 客戶端發(fā)送請求，web server接收請求，然后就是處理請求，最后響應給客戶端這樣一個順序的邏輯。如下圖所示：
代碼實現(xiàn)如下：
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”的頁面來。
通常情況下，大家在開發(fā)web系統(tǒng)的時候，一般都是這么處理請求。接下來我們看在高并發(fā)下如何實現(xiàn)讓請求進行排隊處理。
二、http請求的異步處理方式--排隊處理
讓http請求進入到隊列，我們也稱為異步處理方式。其基本思想就是將接收到的請求的上下文（即request和response）以及處理邏輯包裝成一個工作單元，然后將其放到隊列，然后該工作單元等待消費的工作線程處理該job，處理完成后再返回給客戶端。 流程如下圖：
該實現(xiàn)中會有三個關鍵的元素：工作執(zhí)行單元、隊列、消費者。下面我們逐一看下各自的職責及實現(xiàn)。
工作單元
該工作單元主要是封裝請求的上下文信息（request和response）、請求的處理邏輯以及該工作單元是否被執(zhí)行完成的狀態(tài)。
請求的處理邏輯實際上就是原來在順序處理流程中的具體函數(shù)，如果是mvc模式的話就是controller里的一個具體的action。
在Go中實現(xiàn)通信的方式一般是使用通道。所以，在工作單元中有一個通道，當該工作單元執(zhí)行完具體的處理邏輯后，就往該通道中寫入一個消息，以通知主協(xié)程該次請求已完成，可以返回給客戶端了。
所以，一個http請求的處理邏輯看起來就像是下面這樣：
func (h *MyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
將w和r包裝成工作單元job
將job入隊
等待job執(zhí)行完成
本次請求處理完畢
}
下面我們看下工作單元的具體實現(xiàn)，這里我們將其定義為一個Job結(jié)構(gòu)體：
type Job struct {
DoneChan chan struct{}
handleJob func(j FlowJob) error //具體的處理邏輯
}
Job結(jié)構(gòu)體中有一個handleJob，其類型是一個函數(shù)，即處理請求的邏輯部分。DoneChan通道用來讓該單元進行阻塞等待，并當handleJob執(zhí)行完畢后發(fā)送消息通知的。
下面我們再看看該Job的相關行為：
// 消費者從隊列中取出該job時 執(zhí)行具體的處理邏輯
func (job *Job) Execute() error {
fmt.Println("job start to execute ")
return job.handleJob(job)
}
// 執(zhí)行完Execute后，調(diào)用該函數(shù)以通知主線程中等待的job
func (job *Job) Done() {
job.DoneChan <- struct{}{}
close(job.DoneChan)
}
// 工作單元等待自己被消費
func (job *Job) WaitDone() {
select {
case <-job.DoneChan:
return
}
}
隊列
隊列主要是用來存儲工作單元的。是處理請求的主協(xié)程和消費協(xié)程之間的紐帶。隊列具有列表、容量、當前元素個數(shù)等關鍵元素組成。如下：
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)
}
}
// 消費線程監(jiān)聽隊列的該通道，查看是否有新的job需要消費
func (q *JobQueue) waitJob() <-chan struct{} {
return q.noticeChan
}
這里我們主要解釋一下入隊的操作流程：
- 1 首先是隊列的元素個數(shù)size++
- 2 判斷size是否超過最大容量capacity
- 3 若超過最大容量，則將隊列中最后一個元素移除。因為該元素等待時間最長，認為是超時的情況。
- 4 將新接收的工作單元放入到隊尾。
- 5 往noticeChan通道中寫入一個消息，以便通知消費協(xié)程處理Job。
由以上可知，noticeChan是隊列和消費者協(xié)程之間的紐帶。下面我們來看看消費者的實現(xiàn)。
消費者協(xié)程
消費者協(xié)程的職責是監(jiān)聽隊列，并從隊列中獲取工作單元，執(zhí)行工作單元的具體處理邏輯。在實際應用中，可以根據(jù)系統(tǒng)的承載能力啟用多個消費協(xié)程。在本文中，為了方便講解，我們只啟用一個消費協(xié)程。
我們定義一個WorkerManager結(jié)構(gòu)體，負責管理具體的消費協(xié)程。該WorkerManager有一個屬性是工作隊列，所有啟動的消費協(xié)程都需要從該工作隊列中獲取工作單元。代碼實現(xiàn)如下：
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循環(huán)，然后通過select監(jiān)聽隊列的noticeChan通道，當獲取到工作單元時，就執(zhí)行工作單元中的handleJob方法。執(zhí)行完后，通過job.Done()方法通知在主協(xié)程中還等待的job。這樣整個流程就形成了閉環(huán)。
完整代碼
我們現(xiàn)在看下整體的處理流程，如下圖：
現(xiàn)在我們寫一個測試demo。在這里我們定義了一個全局的flowControl結(jié)構(gòu)體，以作為隊列和工作協(xié)程的管理。代碼如下：
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")
}
之前有一篇文章是優(yōu)先級隊列，實際上就是該隊列的高級實現(xiàn)版本，可以將不同的請求按優(yōu)先級分配到不同的隊列中。有興趣的同學可參考：Go實戰(zhàn) 單隊列到優(yōu)先級隊列的實現(xiàn)
總結(jié)
通過將請求的上下文信息封裝到一個工作單元中，并將其放入到隊列中，然后通過消息通道的方式阻塞等待消費者執(zhí)行完畢。同時在隊列中通過設置隊列的容量以解決請求過多而給系統(tǒng)造成壓力的問題。
以上就是Go http請求排隊處理實戰(zhàn)的詳細內(nèi)容，更多關于Go http請求排隊的資料請關注腳本之家其它相關文章！
相關文章
Golang-如何判斷一個?interface{}?的值是否為?nil
interface?的內(nèi)部實現(xiàn)包含了兩個字段，一個是?type，一個是?data，這篇文章主要介紹了Golang-如何判斷一個interface{}的值是否為nil,需要的朋友可以參考下2023-05-05
在?Go?語言中接口包含兩種含義:它既是方法的集合,?同時還是一種類型并且在Go?語言中是隱式實現(xiàn)的。本文通過示例詳細介紹了Golang接口的使用，需要的可以參考一下2022-09-09
這篇文章主要介紹了談談對Golang IO讀寫的困惑，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-08-08
這篇文章主要介紹了Golang?配置管理庫?Viper,使用?viper?能夠很好的去管理你的配置文件信息，比如數(shù)據(jù)庫的賬號密碼，服務器監(jiān)聽的端口，你可以通過更改配置文件去更改這些內(nèi)容，而不用定位到那一段代碼上去，提高了開發(fā)效率，需要的朋友可以參考下2022-05-05
golang?防緩存擊穿singleflight的實現(xiàn)
本文主要介紹了golang?防緩存擊穿singleflight的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2022-08-08
相信每位編程開發(fā)者們應該都知道，Golang作為一門標榜工程化的語言，提供了非常簡便、實用的編寫單元測試的能力。本文通過Golang源碼包中的用法，來學習在實際項目中如何編寫可測試的Go代碼。有需要的朋友們可以參考借鑒，下面跟著小編一起去學習學習吧。2016-11-11
本文主要介紹了Go語言中sync.Cond使用詳解，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-09-09
切片長度與容量在Go中很常見,切片長度是切片中可用元素的數(shù)量,而切片容量是從切片中第一個元素開始計算的底層數(shù)組中的元素數(shù)量,這篇文章主要給大家介紹了關于Golang中切片長度和容量區(qū)別的相關資料,需要的朋友可以參考下2024-01-01
go語言使用Chromedp實現(xiàn)二維碼登陸教程示例源碼
這篇文章主要為大家介紹了go語言使用Chromedp實現(xiàn)二維碼登陸示例源碼，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-04-04
這篇文章主要為大家介紹了Go語言kafka生產(chǎn)消費消息的實例搬磚，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-06-06