Python并發(fā)請(qǐng)求下如何實(shí)現(xiàn)限制QPS

這篇文章主要為大家展示了Python并發(fā)請(qǐng)求下如何實(shí)現(xiàn)限制QPS，內(nèi)容簡(jiǎn)而易懂，希望大家可以學(xué)習(xí)一下，學(xué)習(xí)完之后肯定會(huì)有收獲的，下面讓小編帶大家一起來(lái)看看吧。

前兩天有一個(gè)需求，需要訪問(wèn)某API服務(wù)器請(qǐng)求數(shù)據(jù)，該服務(wù)器限制了QPS=2，因?yàn)镼PS很小所以就使用阻塞式請(qǐng)求。后來(lái)開(kāi)通了服務(wù)，QPS提高到了20，阻塞式請(qǐng)求滿足不了這個(gè)QPS了，于是使用了GRequests來(lái)并發(fā)請(qǐng)求數(shù)據(jù)，但這里又遇到了一個(gè)問(wèn)題：并發(fā)太快，服務(wù)器通過(guò)發(fā)送錯(cuò)誤碼拒絕了很多數(shù)據(jù)的響應(yīng)，造成了資源的浪費(fèi)。
故在此記錄以下幾種 節(jié)流(Throttle) 方法：

以下均假設(shè)有如下包和數(shù)據(jù)前提：

import grequests

urls = [
 "https://www.baidu.com",
 "https://www.google.com"
]
requests = [
 grequests.get(url)
 for url in urls
] * 1000

rate = 20 # 表示 20 請(qǐng)求/秒

time.sleep(1)

這是最簡(jiǎn)單的方法，通過(guò)time.sleep(1)阻塞進(jìn)程來(lái)控制每秒并發(fā)數(shù)量。用公式表達(dá)如下：Time=請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延+time.sleep(1)Time = 請(qǐng)求準(zhǔn)備時(shí)延 + 請(qǐng)求發(fā)送時(shí)延 + time.sleep(1)Time=請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延+time.sleep(1) 但是這種方法有一個(gè)較小的問(wèn)題：不精確 。數(shù)據(jù)量越大，方差越大。

from time import sleep

req_groups = [
 requests[i: i+rate]
 for i in range(0, len(requests), rate)
]

ret = []
for req_group in req_groups:
 ret += grequests.map(req_group)
 sleep(1)

print(ret)

令牌桶(token bucket)方法

這種方法較精確，可以確保誤差不超過(guò)±1（當(dāng)然前提是你的電腦和目標(biāo)服務(wù)器都能承受的了高并發(fā)）。以下是耗時(shí)的公式表示：Time=請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延+令牌桶阻塞時(shí)延Time = 請(qǐng)求準(zhǔn)備時(shí)延 + 請(qǐng)求發(fā)送時(shí)延 + 令牌桶阻塞時(shí)延Time=請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延+令牌桶阻塞時(shí)延令牌桶阻塞時(shí)延≈1−請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延令牌桶阻塞時(shí)延 ≈ 1 - 請(qǐng)求準(zhǔn)備時(shí)延 + 請(qǐng)求發(fā)送時(shí)延令牌桶阻塞時(shí)延請(qǐng)求準(zhǔn)備時(shí)延+請(qǐng)求發(fā)送時(shí)延 這種方法當(dāng)然也有一點(diǎn)缺陷，CPU看起來(lái)會(huì)很高（這是由于 while pass），盡管CPU真實(shí)使用率很低。

from time import time

class Throttle:
 def __init__(self, rate):
  self.rate = rate
  self.tokens = 0
  self.last = 0
 
 def consume(self, amount=1):
  now = time()
  
  if self.last == 0:
   self.last = now
  
  elapsed = now - self.last

  if int(elapsed * self.rate):
   self.tokens += int(elapsed * self.rate)
   self.last = now
  
  self.tokens = (
   self.rate
   if self.tokens > self.rate
   else self.tokens
  )
  
  if self.tokens >= amount:
   self.tokens -= amount
  else:
   amount = 0
  
  return amount

throttle = Throttle(rate)

req_groups = [
 requests[i: i+rate]
 for i in range(0, len(requests), rate)
]

ret = []
for req_group in req_groups:
 ret += grequests.map(req_group)
 while throttle.consume():
  pass # 阻塞

print(ret)

GRequests-Throttle

這是一個(gè)使用令牌桶(token bucket)方法進(jìn)行封裝的GRequests修改版，使用方法很簡(jiǎn)單：
首先安裝grequests-throttle(清華鏡像源更新較慢，推薦使用阿里鏡像源)

pip install grequests-throttle

import grequests_throttle as gt

ret = gt.map(requests, rate=rate)
print(ret)

以上就是關(guān)于Python并發(fā)請(qǐng)求下如何實(shí)現(xiàn)限制QPS的內(nèi)容，如果你們有學(xué)習(xí)到知識(shí)或者技能，可以把它分享出去讓更多的人看到。