前兩天有一個需求，需要訪問某API服務(wù)器請求數(shù)據(jù)，該服務(wù)器限制了QPS=2(哈哈應(yīng)該都知道是哪個服務(wù)器了吧_(:з」∠)_)，因?yàn)镼PS很小所以就使用阻塞式請求。后來開通了服務(wù)，QPS提高到了20，阻塞式請求滿足不了這個QPS了，于是使用了GRequests來并發(fā)請求數(shù)據(jù)，但這里又遇到了一個問題：并發(fā)太快，服務(wù)器通過發(fā)送錯誤碼拒絕了很多數(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 請求/秒

time.sleep(1)

  這是最簡單的方法，通過time.sleep(1)阻塞進(jìn)程來控制每秒并發(fā)數(shù)量。用公式表達(dá)如下：$$Time=請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延+time.sleep(1)Time = 請求準(zhǔn)備時(shí)延 + 請求發(fā)送時(shí)延 + time.sleep(1)Time=請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延+time.sleep(1)   但是這種方法有一個較小的問題：不精確 。數(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)方法

  這種方法較精確，可以確保誤差不超過±1（當(dāng)然前提是你的電腦和目標(biāo)服務(wù)器都能承受的了高并發(fā)）。以下是耗時(shí)的公式表示：$$Time=請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延+令牌桶阻塞時(shí)延Time = 請求準(zhǔn)備時(shí)延 + 請求發(fā)送時(shí)延 + 令牌桶阻塞時(shí)延Time=請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延+令牌桶阻塞時(shí)延 $$令牌桶阻塞時(shí)延≈1−請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延令牌桶阻塞時(shí)延 ≈ 1 - 請求準(zhǔn)備時(shí)延 + 請求發(fā)送時(shí)延令牌桶阻塞時(shí)延≈1−請求準(zhǔn)備時(shí)延+請求發(fā)送時(shí)延   這種方法當(dāng)然也有一點(diǎn)缺陷，CPU看起來會很高（這是由于 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

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

pip install grequests-throttle

import grequests_throttle as gt

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

總結(jié)

  如果并發(fā)請求數(shù)量較小，可以考慮使用time.sleep(1)簡單快捷；當(dāng)并發(fā)請求數(shù)量較大時(shí)，使用令牌桶(token bucket)方法能最大化利用每一秒；如果不想寫太多代碼，可以使用GRequests-Throttle包進(jìn)行請求流量控制。

到此這篇關(guān)于Python并發(fā)請求下限制QPS(每秒查詢率)實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Python并發(fā)請求下限制QPS(每秒查詢率)實(shí)現(xiàn)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論