作者：林冠巨集 / 指尖下的幽靈

部落格：http://www.cnblogs.com/linguanh/

GitHub ： https://github.com/af913337456/

---

目錄

• 前序
• 一般的訂單流程
• 思考瓶頸點
• 訂單佇列
  • 第一種訂單佇列
    
  • 第二種訂單佇列
  • 總結
• 實現佇列的選擇
• 解答
• 實現佇列的選擇
• 第二種佇列的 Go 版本例子程式碼

前序

目前的開發工作主要是將傳統電商應用和區塊鏈技術相結合，區塊鏈平臺依然是以太坊，此外地，這幾天由我編寫，經清華大學出版社出版的書籍，歷經八月，終於出版上架了，名稱是：《區塊鏈以太坊DApp開發實戰》，現已可以網購。

本文所要分享的思路就是電商應用中常用的訂單佇列

。

一般的訂單流程

電商應用中，簡單直觀的使用者從下單到付款，最終完成整個流程的步驟可以用下圖表示：

其中，訂單資訊持久化，就是儲存資料到資料庫中。而最終客戶端完成支付後的更新訂單狀態的操作是由第三方支付平臺進行回撥設定好的回撥連結 NotifyUrl，來進行的。

補全訂單狀態的更新流程，如下圖表示：

思考瓶頸點

服務端的直接瓶頸點，首先要考慮 TPS。去除細分點，我們主要看訂單資訊持久化瓶頸點。

在高併發業務場景中，例如 秒殺、優惠價搶購等。短時間內的下單請求數會很多，如果訂單資訊持久化 部分，不做優化，而是直接對資料庫層進行頻繁的讀寫操作，資料庫會承受不了，容易成為第一個垮掉的服務，比如下圖的所示的常規寫單流程：

可以看到，每持久化一個訂單資訊，一般要經歷網路連線操作（連結資料庫），以及多個 I/O 操作。

得益於連線池技術，我們可以在連結資料庫的時候，不用每次都重新發起一次完整的HTTP請求，而可以直接從池中獲取已打開了的連線控制代碼，而直接使用，這點和執行緒池的原理差不多。

此外，我們還可以在上面的流程中加入更多的優化，例如對於一些需要讀取的資訊，可以事先存置到記憶體快取層，並加於更新維護，這樣在使用的時候，可以快速讀取。

即使我們都具備了上述的一些優化手段，但是對於寫操作的I/O阻塞耗時，在高併發請求的時候，依然容易導致資料庫承受不住，容易出現連結多開異常，操作超時等問題。

在該層進行優化的操作，除了上面談到的之外，還有下面一些手段：

• 資料庫叢集，採用讀寫分離，減少寫時壓力
• 分庫，不同業務的表放到不同的資料庫，會引入分散式事務問題
• 採用佇列模型削峰

每種方式有各自的特點，因為本文談的是訂單佇列的架構思想，所以下面我們來看下如何在訂單系統中引入訂單佇列。

訂單佇列

網上有不少文章談到訂單佇列的做法，大部分都漏了說明請求與響應的一致性問題。

第一種訂單佇列流程圖：

上圖是大多文章提到的佇列模型，有兩個沒有解析的問題：

1. 如果訂單存在第三方支付情況，① 和 ② 的一致性如何保證，比如其中一處處理失敗；
2. 如果訂單存在第三方支付情況，① 完成了支付，且三方支付平臺回調了 notifyUrl，而此時 ② 還在排隊等待處理，這種情況又如何處理。

首先，要肯定的是，上面的訂單流程圖是沒有問題的。它有下面的優缺點，所提到的兩個問題也是有解決方案的。

優點：

• 使用者無需等待訂單持久化處理，而能直接獲得響應，實現快速下單
• 持久化處理，採用排隊的先來先處理，不會像上面談到的高併發請求一起衝擊資料庫層面的情況。
• 可變性強，搭配中介軟體的組合性強。

缺點：

• 多訂單入隊時，② 步驟的處理速度跟不上。從而導致第二點問題。
• 實現較複雜

上面談及的問題點，我後面都會給出解決方案。下面我們來看下另外一種訂單佇列流程圖。

第二種訂單佇列流程圖：

第二種訂單佇列的設計模型，注意它的同步等待持久化處理的結果，解決了持久化與響應的一致性問題，但是有個嚴重的耗時等待問題，它的優缺點如下：

優點：

1. 持久化與響應的強一致性。
2. 持久化處理，採用排隊的先來先處理，不會像上面談到的高併發請求一起衝擊資料庫層面的情況。
3. 實現簡單

缺點：

1. 多訂單入隊時，持久化單元處理速度跟不上，造成客戶端同步等待響應。

這類訂單佇列，我下面會放出 Golang 實現的版本程式碼。

總結

對比上面兩種常見的訂單模型，如果從使用者體驗的角度去優先考慮，第一種不需要使用者等待持久化處理結果的是明顯優於第二種的。如果技術團隊完善，且技術過硬，也應該考慮第一種的實現方式。

如果僅僅想要達到寧願使用者等待到超時也不願意儲存層服務被沖垮，那麼有限考慮第二種。

實現佇列的選擇

在這裡，我們進一步細分一下，實現佇列模組的功能有哪些選擇。

相信很多後端開發經驗比較老道的同志已經想到了，使用現有的中介軟體，比如知名的 Redis、RocketMQ，以及 Kafka 等，它們都是一種選擇。

此外地，我們還可以直接編寫程式碼，在當前的服務系統中實現一個訊息佇列來達到目的，下面我用圖來分類下佇列型別。

不同的佇列實現方式，能直接導致不同的功能，也有不同的優缺點：

一級快取優點：

1. 一級快取，最快。無需連結，直接從記憶體層獲取；
2. 如果不考慮持久化和叢集，那麼它實現簡單。

一級快取缺點：

1. 如果考慮持久化和叢集，那麼它實現比較複雜。
2. 不考慮持久化情況下，如果伺服器斷電或其它原因導致服務中斷，那麼排隊中的訂單資訊將丟失

中介軟體的優點：

1. 軟體成熟，一般出名的訊息中介軟體都是經過實踐使用的，文件豐富；
2. 支援多種持久化的策略，比如 Redis 有增量持久化，能最大程度減少因不可預料的崩潰導致訂單資訊丟失；
3. 支援叢集，主從同步，這對於分散式系統來說，是必不可少的要求。

中介軟體的缺點：

1. 分散式部署時，需要建立連結通訊，導致讀寫操作需要走網路通訊。

解答

回到第一種訂單模型中：

問題1：

如果訂單存在第三方支付情況，① 和 ② 的一致性如何保證？

首先我們看下，不一致性的時候，會產生什麼結果：

1. ① 失敗，使用者因為網路原因或返回其它頁面，不能獲取結果。而 ② 成功，那麼最終該訂單的狀態是待支付。使用者進入到個人訂單中心完成訂單支付即可；
2. ① 和 ② 都失敗，那麼下單失敗；
3. ① 成功，② 失敗，此時使用者在響應頁面完成了支付動作，使用者檢視訂單資訊為空白。

上述的情況，明顯地，只有 3 是需要恢復訂單資訊的，應對的方案有：

• 當服務端支付回撥介面被第三方支付平臺訪問時，無法找到對應的訂單資訊。那麼先將這類支付了卻沒訂單資訊的資料儲存起來先，比如儲存到表A。同時啟動一個定時任務B專門遍歷表A，然後去訂單列表尋找是否已經有了對應的訂單資訊，有則更新，沒則繼續，或跟隨制定的檢測策略走。
• 當 ② 是由於服務端的非崩潰性原因而導致失敗時：
  • 失敗的時候同時將原始訂單資料重新插入到佇列頭部，等待下一次的重新持久化處理。
• 當 ② 因服務端的崩潰性原因而導致失敗時：
  • 定時任務B在進行了多次檢測無果後，那麼根據第三方支付平臺在回撥時候傳遞過來的訂單附屬資訊對訂單進行恢復。
• 整個過程訂單恢復的過程，使用者檢視訂單資訊為空白。
• 定時任務B 所在服務最好和回撥連結 notifyUrl 所在的介面服務一致，這樣能保證當 B 掛掉的時候，回撥服務也跟隨掛掉，然後第三方支付平臺在呼叫回撥失敗的情況下，他們會有重試邏輯，依賴這個，在回撥服務重啟時，可以完成訂單資訊恢復。

問題2：

如果訂單存在第三方支付情況，① 完成了支付，且三方支付平臺回調了 notifyUrl，而此時 ② 還在排隊等待處理，這種情況又如何處理？

應對的方案參考 問題1 的 定時任務B 檢測修改機制。

第二種佇列的 Go 版本例子程式碼

定義一些常量

const (
    QueueOrderKey   = "order_queue"     
    QueueBufferSize = 1024              // 請求佇列大小
    QueueHandleTime = time.Second * 7   // 單個 mission 超時時間
)

定義出入隊介面，方便多種實現

// 定義出入隊介面，方便多種實現
type IQueue interface {
    Push(key string,data []byte) error
    Pop(key string) ([]byte,error)
}

定義請求與響應實體

// 定義請求與響應實體
type QueueTimeoutResp struct {
    Timeout  bool  // 超時標誌位
    Response chan interface{}
}
type QueueRequest struct {
    ReqId       string  `json:"req_id"`  // 單次請求 id
    Order       *model.OrderCombine `json:"order"` // 訂單資訊 bean
    AccessTime  int64   `json:"access_time"` // 請求時間
    ResponseChan *QueueTimeoutResp `json:"-"`
}

定義佇列實體

// 定義佇列實體
type Queue struct {
    mapLock sync.Mutex
    RequestChan  chan *QueueRequest // 快取管道，裝載請求
    RequestMap   map[string]*QueueTimeoutResp 
    Queue IQueue
}

例項化佇列，接收介面引數

// 例項化佇列，接收介面引數
func NewQueue(queue IQueue) *Queue {
    return &Queue{
        mapLock:     sync.Mutex{},
        RequestChan: make(chan *QueueRequest, QueueBufferSize),
        RequestMap:  make(map[string]*QueueTimeoutResp, QueueBufferSize),
        Queue:       queue,
    }
}

接收請求

// 接收請求
func (q *Queue) AcceptRequest(req *QueueRequest) interface{} {
    if req.ResponseChan == nil {
        req.ResponseChan = &QueueTimeoutResp{
            Timeout:  false,
            Response: make(chan interface{},1),
        }
    }
    userKey := key(req)  // 唯一 key 生成函式
    req.ReqId = userKey
    q.mapLock.Lock()
    q.RequestMap[userKey] = req.ResponseChan // 記憶體層儲存對應的 req 的 resp 管道指標
    q.mapLock.Unlock()
    q.RequestChan <- req  // 接收請求
    log("userKey : ", userKey)
    ticker := time.NewTicker(QueueHandleTime) // 以超時時間 QueueHandleTime 啟動一個定時器
    defer func() {
        ticker.Stop() // 釋放定時器
        q.mapLock.Lock()
        delete(q.RequestMap,userKey)  // 當處理完一個 req，從 map 中移出
        q.mapLock.Unlock()
    }()

    select {
    case <-ticker.C:  // 超時
        req.ResponseChan.Timeout = true 
        Queue_TimeoutCounter++  // 輔助計數，int 型別
        log("timeout: ",userKey)
        return lghError.HandleTimeOut  // 返回超時錯誤的資訊
    case result := <-req.ResponseChan.Response:  // req 被完整處理
        return result
    }
}

從請求管道中取出 req 放入到佇列容器中，該函式在 gorutine 中執行

// 從請求管道中取出 req 放入到佇列容器中，該函式在 gorutine 中執行
func (q *Queue) addToQueue() {
    for {
        req := <-q.RequestChan // 取出一個 req
        data, err := json.Marshal(req)
        if err != nil {
            log("redis queue parse req failed : ", err.Error())
            continue
        }
        if err = q.Queue.Push(QueueOrderKey, data);err != nil {  // push 入隊，這裡有時間消耗
            log("lpush req failed. Error : ", err.Error())
            continue
        }
        log("lpush req success. req time: ", req.AccessTime)
    }
}

取出 req 處理，該函式在 gorutine 中執行

// 取出 req 處理，該函式在 gorutine 中執行
func (q *Queue) readFromQueue() {
    for {
        data, err := q.Queue.Pop(QueueOrderKey) // pop 出隊，這裡也有時間消耗
        if err != nil {
            log("lpop failed. Error : ", err.Error())
            continue
        }
        if data == nil || len(data) == 0 {
            time.Sleep(time.Millisecond * 100) // 空資料的 req，停頓下再取
            continue
        }
        req := &QueueRequest{}
        if err = json.Unmarshal(data, req);err != nil {
            log("Lpop: json.Unmarshal failed. Error : ", err.Error())
            continue
        }
        userKey := req.ReqId
        q.mapLock.Lock()
        resultChan, ok := q.RequestMap[userKey] // 取出對應的 resp 管道指標
        q.mapLock.Unlock()
        if !ok {
            // 中介軟體重啟時，比如 redis 重啟而讀取舊 key，會進入這裡
            Queue_KeyNotFound ++ // 計數 int 型別
            log("key not found, rollback: ", userKey)
            continue
        }
        simulationTimeOutReq4(req) // 模擬出來任務的函式，入參為 req 
        if resultChan.Timeout {
            // 處理期間，已經超時，這裡做可以拓展回滾操作
            Queue_MissionTimeout ++
            log("handle mission timeout: ", userKey)
            continue
        }
        log("request result send to chan succeee, userKey : ", userKey)
        ret := util.GetCommonSuccess(req.AccessTime)
        resultChan.Response <- &ret // 輸入處理成功
    }
}

啟動

func (q *Queue) Start()  {
    go q.addToQueue()
    go q.readFromQueue()
}

執行例子

func test(){
    ...
    runtime.GOMAXPROCS(4)
    redisQueue := NewQueue(NewFastCacheQueue())
    redisQueue.Start()
    reqNumber := testReqNumber
    wg := sync.WaitGroup{}
    wg.Add(reqNumber)
    for i :=0;i<reqNumber;i++ {
        go func(index int) {
            combine := model.OrderCombine{}
            ret := AcceptRequest(&QueueRequest{
                UserId:       int64(index),
                Order:        &combine,
                AccessTime:   time.Now().Unix(),
                ResponseChan: nil,
            })
            fmt.Println("ret: ------------- ",ret.String())
            wg.Done()
        }(i)
    }
    wg.Wait()
    time.Sleep(3*time.Second)
    fmt.Println("TimeoutCounter: ",Queue_TimeoutCounter,"KeyNotFound: ",Queue_KeyNotFound,"MissionTimeout: ",Queue_MissionTimeout)
}

最後上傳一張書籍圖片

相關推薦