golang API 請求隊列

• 概要
• 實現思路
• 使用方法
  • 啓動隊列服務
  • 使用隊列服務

概要

在調用第三方 API 的時候, 基本都有訪問限速的限制條件. 第三方的 API 有多個的時候, 就不太好控制訪問速度, 經常會致使 HTTP 429(Too Many Requests) 而後就會有一段時間的禁止訪問.

爲了應對這種限速的狀況, 經過一個簡單的請求隊列來控制訪問的速度, 以後基本沒遇到過 HTTP 429 了.

實現思路

首先, 每一個請求包裝成一個 RequestParam 的 struct, 其中包含請求的地址,類型,參數以及 response 的 channel.

發送請求的時候, 只要將 RequestParam 放入請求隊列中便可, 請求完成後, 將 response 放入對應的 channel 中.

整個代碼實現很簡單:

1  package util
 2  
 3  import (
 4    "fmt"
 5  
 6    apiclient "gitee.com/wangyubin/gutils/api_client"
 7    "gitee.com/wangyubin/gutils/logger"
 8  )
 9  
10  // request 包含的內容
11  type RequestParam struct {
12    Api     string
13    Method  string
14    JsonReq interface{}
15    Resp    chan []byte
16  }
17  
18  // 請求隊列, 本質是一個channel
19  type RequestQueue struct {
20    Queue chan RequestParam
21  }
22  
23  var queue *RequestQueue
24  
25  // 獲取隊列
26  func GetQueue() *RequestQueue {
27    return queue
28  }
29  
30  // 初始化隊列
31  func InitRequestQueue(size int) {
32    queue = &RequestQueue{
33      Queue: make(chan RequestParam, size),
34    }
35  }
36  
37  // 將請求放入隊列
38  func (rq *RequestQueue) Enqueue(p RequestParam) {
39    rq.Queue <- p
40  }
41  
42  // 請求隊列服務, 一直等待接受和處理請求
43  func (rq *RequestQueue) Run() {
44    lg := logger.GetLogger()
45    for p := range rq.Queue {
46      var resp []byte
47      var err error
48      switch p.Method {
49      case "GET":
50        resp, err = apiclient.GetJson(p.Api, p.JsonReq)
51      case "POST":
52        resp, err = apiclient.PostJson(p.Api, p.JsonReq)
53      default:
54        err = fmt.Errorf("Wrong type of METHOD(%s)\n", p.Method)
55      }
56  
57      if err != nil {
58        lg.Err(err).Msg("access api error: " + p.Api)
59        continue
60      }
61      if p.Resp != nil {
62        p.Resp <- resp
63        close(p.Resp)
64      }
65    }
66  
67    lg.Info().Msg("request queue finished!")
68  }

這裏的請求是用了我本身封裝的 apiclient, 能夠根據實際狀況替換.

在個人應用場景裏, 只要 api 順序訪問就不會出現 HTTP 429 了, 若是這樣以爲速度太快的的話, 能夠嘗試在 Run() 函數中加入一些時間間隔.

1  func (rq *RequestQueue) Run() {
2    lg := logger.GetLogger()
3    for p := range rq.Queue {
4       time.Sleep(1 * time.Second)
5       // ... 省略的代碼 ...
6    }
7  
8    lg.Info().Msg("request queue finished!")
9  }

使用方法

使用很簡單, 首先啓動, 而後每一個調用的地方將 RequestParam 放入隊列並等待 response 便可.

啓動隊列服務

1  func main() {
2      // init request queue and start queue service
3      util.InitRequestQueue(100)
4      queue := util.GetQueue()
5      defer close(queue.Queue)
6      go queue.Run()
7  
8      // 其餘啓動代碼
9  }

使用隊列服務

1  func Request(param1 string, param2 int) error {
 2   api := "http://xxxx.com"
 3   api = fmt.Sprintf("%s?period=%s&size=%d", api, param1, param2)
 4  
 5   queue := util.GetQueue()
 6   param := util.RequestParam{
 7     Api:    api,
 8     Method: "GET",
 9     Resp:   make(chan []byte, 1),
10   }
11   queue.Enqueue(param)
12  
13   var respData struct {
14     Status string       `json:"status"`
15     Data   []model.Data `json:"data"`
16   }
17   var err error
18   for resp := range param.Resp {
19     err = json.Unmarshal(resp, &respData)
20     if err != nil {
21       lg.Err(err).Msg("unmarshal json error")
22       return err
23     }
24   }
25  
26   fmt.Println(respData) 
27   return  err
28  }