前兩天有一個需求，需要訪問某API服務器請求數據，該服務器限制了QPS=2(哈哈應該都知道是哪個服務器了吧_(:з」∠)_)，因為QPS很小所以就使用阻塞式請求。后來開通了服務，QPS提高到了20，阻塞式請求滿足不了這個QPS了，于是使用了GRequests來并發請求數據，但這里又遇到了一個問題：并發太快，服務器通過發送錯誤碼拒絕了很多數據的響應，造成了資源的浪費。

故在此記錄以下幾種 節流(Throttle) 方法：

以下均假設有如下包和數據前提：

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)阻塞進程來控制每秒并發數量。用公式表達如下：Time=請求準備時延+請求發送時延+time.sleep(1)Time = 請求準備時延 + 請求發送時延 + time.sleep(1)Time=請求準備時延+請求發送時延+time.sleep(1) ??但是這種方法有一個較小的問題：不精確 。數據量越大，方差越大。

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（當然前提是你的電腦和目標服務器都能承受的了高并發）。以下是耗時的公式表示：Time=請求準備時延+請求發送時延+令牌桶阻塞時延Time = 請求準備時延 + 請求發送時延 + 令牌桶阻塞時延Time=請求準備時延+請求發送時延+令牌桶阻塞時延 令牌桶阻塞時延≈1?請求準備時延+請求發送時延令牌桶阻塞時延 ≈ 1 - 請求準備時延 + 請求發送時延令牌桶阻塞時延≈1?請求準備時延+請求發送時延 ??這種方法當然也有一點缺陷，CPU看起來會很高（這是由于 while pass），盡管CPU真實使用率很低。

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)方法進行封裝的GRequests修改版，使用方法很簡單：

首先安裝grequests-throttle(清華鏡像源更新較慢，推薦使用阿里鏡像源)

pip install grequests-throttle

import grequests_throttle as gt

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

print(ret)

總結

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

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

總結

以上是生活随笔為你收集整理的python测试程序的qps和响应时间代码_Python并发请求下限制QPS(每秒查询率)的实现代码...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。