python腳本請(qǐng)求數(shù)量達(dá)到上限,http請(qǐng)求重試問(wèn)題

更新時(shí)間：2024年06月27日 09:54:05 作者：拾牙慧者

這篇文章主要介紹了python腳本請(qǐng)求數(shù)量達(dá)到上限,http請(qǐng)求重試問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

python請(qǐng)求數(shù)量達(dá)到上限,http請(qǐng)求重試

由于在內(nèi)網(wǎng)發(fā)送http請(qǐng)求同一個(gè)token會(huì)限制次數(shù)，所以很容易達(dá)到網(wǎng)關(guān)流量上限。

業(yè)務(wù)中使用了多線程并發(fā)，一個(gè)線程發(fā)起一次http請(qǐng)求，得到正確結(jié)果后返回。

這里采用的策略是，如果解析出來(lái)達(dá)到流量上限，那么該線程休眠一段時(shí)間，然后重試請(qǐng)求，如果還是失敗，那么繼續(xù)休眠，每次休眠的時(shí)間隨著重試輪次增加：

# 探測(cè)是否觸及網(wǎng)關(guān)流量上限
def probe_func(m_url, m_headers, m_json, m_timeout):
    json_rep = requests.post(url = m_url, 
                    headers = m_headers,
                    json = m_json,
                    timeout = m_timeout)
    zhiyan_data = json_rep.json()
    if(zhiyan_data['code'] != 0):
        return None
    else:
        return json_rep

# 解析數(shù)據(jù)包，不涉及probe_func中的檢測(cè)內(nèi)容
def parse(json_rep, room_name, metric_name):
    if json_rep == None: 
        logging.info(room_name + " json_rep == None")
        return 0
    if (json_rep.content and json_rep.status_code != 204 and json_rep.headers["content-type"].strip().startswith("application/json")):
        zhiyan_data = json_rep.json()
        if len(zhiyan_data['data']) == 0:
            logging.warning(zhiyan_data['日志信息拉取無(wú)結(jié)果'])
            return 0
        else:
            res = zhiyan_data['data']['chart_info'][0]['key_data_list'][3]['current']
            logging.info(room_name + str(res))
            if str(res) == "None":
                logging.warning(room_name + ":拉取zhiyan_data:" + metric_name + " 出現(xiàn)了問(wèn)題，拉取數(shù)據(jù)為None")
                return 0
            else:
                return res
    else:
        return 0

# 具有可靠性地獲取數(shù)據(jù)
def request_post_reliable(m_url, m_headers, m_json, m_timeout):
    sleep_time_s = 1
    sleep_time_max = 60
    res = probe_func(m_url, m_headers, m_json, m_timeout)
    # 如果探測(cè)失敗則線程睡眠一段時(shí)間后再嘗試
    while (res == None):
        logging.info("探測(cè)失敗，線程睡眠"+str(sleep_time_s)+"秒")
        time.sleep(sleep_time_s)
        tmp = sleep_time_s * 2
        if tmp < sleep_time_max:
            sleep_time_s = tmp
        else:
            sleep_time_s = sleep_time_max
        logging.info("睡眠結(jié)束，線程重新探測(cè)")
        res = probe_func(m_url, m_headers, m_json, m_timeout)
    # 直到探測(cè)成功，返回正確結(jié)果
    return res

python請(qǐng)求http/https時(shí)設(shè)置失敗重試次數(shù)

使用Python的requests庫(kù)時(shí)，默認(rèn)是沒(méi)有失敗時(shí)重試請(qǐng)求的，通過(guò)下面的方式可以支持重試請(qǐng)求

設(shè)置請(qǐng)求時(shí)的重試規(guī)則

import requests
from requests.adapters import HTTPAdapter

s = requests.Session()
a = HTTPAdapter(max_retries=3)
b = HTTPAdapter(max_retries=3)
#將重試規(guī)則掛載到http和https請(qǐng)求
s.mount('http://', a)
s.mount('https://', b)

請(qǐng)求Url

上面設(shè)置完畢后，通過(guò)改Session的請(qǐng)求就可以支持失敗重試

r = s.get('http://api.map.baidu.com/geocoder?location=39.90733345,116.391244079988&output=json')
# 返回的狀態(tài)碼
r.status_code
# 響應(yīng)內(nèi)容，中文為utf8編碼
r.content
# 響應(yīng)的字符串形式，中文為unicode編碼
r.text
# 響應(yīng)頭中的編碼
r.encoding
# 響應(yīng)頭信息
r.headers