golang API请求队列的实现

概要

在调用第三方 api 的时候, 基本都有访问限速的限制条件. 第三方的 api 有多个的时候, 就不太好控制访问速度, 常常会导致 http 429(too many requests) 然后就会有一段时间的禁止访问.

为了应对这种限速的情况, 通过一个简单的请求队列来控制访问的速度, 之后基本没遇到过 http 429 了.

实现思路

首先, 每个请求包装成一个 requestparam 的 struct, 其中包含请求的地址,类型,参数以及 response 的 channel.

发送请求的时候, 只要将 requestparam 放入请求队列中即可, 请求完成后, 将 response 放入对应的 channel 中.

整个代码实现很简单:

   package util
   
   import (
     "fmt"
   
     apiclient "gitee.com/wangyubin/gutils/api_client"
     "gitee.com/wangyubin/gutils/logger"
   )
   
  // request 包含的内容
  type requestparam struct {
    api     string
    method  string
    jsonreq interface{}
    resp    chan []byte
  }
  
  // 请求队列, 本质是一个channel
  type requestqueue struct {
    queue chan requestparam
  }
  
  var queue *requestqueue
  
  // 获取队列
  func getqueue() *requestqueue {
    return queue
  }
  
  // 初始化队列
  func initrequestqueue(size int) {
    queue = &requestqueue{
      queue: make(chan requestparam, size),
    }
  }
  
  // 将请求放入队列
  func (rq *requestqueue) enqueue(p requestparam) {
    rq.queue <- p
  }
  
  // 请求队列服务, 一直等待接受和处理请求
  func (rq *requestqueue) run() {
    lg := logger.getlogger()
    for p := range rq.queue {
      var resp []byte
      var err error
      switch p.method {
      case "get":
        resp, err = apiclient.getjson(p.api, p.jsonreq)
      case "post":
        resp, err = apiclient.postjson(p.api, p.jsonreq)
      default:
        err = fmt.errorf("wrong type of method(%s)\n", p.method)
      }
  
      if err != nil {
        lg.err(err).msg("access api error: " + p.api)
        continue
      }
      if p.resp != nil {
        p.resp <- resp
        close(p.resp)
      }
    }
  
    lg.info().msg("request queue finished!")
  }

这里的请求是用了我自己封装的 apiclient, 可以根据实际情况替换.

在我的应用场景里, 只要 api 顺序访问就不会出现 http 429 了, 如果这样觉得速度太快的的话, 可以尝试在 run() 函数中加入一些时间间隔.

  func (rq *requestqueue) run() {
    lg := logger.getlogger()
    for p := range rq.queue {
       time.sleep(1 * time.second)
       // ... 省略的代码 ...
    }
  
    lg.info().msg("request queue finished!")
  }

使用方法

使用很简单, 首先启动, 然后每个调用的地方将 requestparam 放入队列并等待 response 即可.

启动队列服务 

func main() {
      // init request queue and start queue service
      util.initrequestqueue(100)
      queue := util.getqueue()
      defer close(queue.queue)
      go queue.run()
  
      // 其他启动代码
  }

使用队列服务

   func request(param1 string, param2 int) error {
   api := "http://xxxx.com"
   api = fmt.sprintf("%s?period=%s&size=%d", api, param1, param2)
   
    queue := util.getqueue()
    param := util.requestparam{
      api:    api,
      method: "get",
     resp:   make(chan []byte, 1),
   }
   queue.enqueue(param)
  
   var respdata struct {
     status string       `json:"status"`
     data   []model.data `json:"data"`
   }
   var err error
   for resp := range param.resp {
     err = json.unmarshal(resp, &respdata)
     if err != nil {
       lg.err(err).msg("unmarshal json error")
       return err
     }
   }
  
   fmt.println(respdata) 
   return  err
  }