Python控制函數運行時間

在某個Flask項目在做后端接口時需要設置超時響應，因為接口中使用爬蟲請求了多個網站,響應時間時長時短。

我需要設置一個最大響應時間，時間內如果接口爬蟲沒跑完,直接返回請求超時。

從網上了解到有兩種方法，廢話不多說直接上代碼。

方法1:使用線程控制

import requests, datetime, time
import threading


class MyThread(threading.Thread):
    def __init__(self, target, args=()):
        """
        why: 因為threading類沒有返回值,因此在此處重新定義MyThread類,使線程擁有返回值
        此方法來源 https://www.cnblogs.com/hujq1029/p/7219163.html?utm_source=itdadao&utm_medium=referral
        """
        super(MyThread, self).__init__()
        self.func = target
        self.args = args

    def run(self):
        # 接受返回值
        self.result = self.func(*self.args)

    def get_result(self):
        # 線程不結束,返回值為None
        try:
            return self.result
        except Exception:
            return None


# 為了限制真實請求時間或函數執行時間的裝飾器
def limit_decor(limit_time):
    """
    :param limit_time: 設置最大允許執行時長,單位:秒
    :return: 未超時返回被裝飾函數返回值,超時則返回 None
    """

    def functions(func):
        # 執行操作
        def run(*params):
            thre_func = MyThread(target=func, args=params)
            # 主線程結束(超出時長),則線程方法結束
            thre_func.setDaemon(True)
            thre_func.start()
            # 計算分段沉睡次數
            sleep_num = int(limit_time // 1)
            sleep_nums = round(limit_time % 1, 1)
            # 多次短暫沉睡並嘗試獲取返回值
            for i in range(sleep_num):
                time.sleep(1)
                infor = thre_func.get_result()
                if infor:
                    return infor
            time.sleep(sleep_nums)
            # 最終返回值(不論線程是否已結束)
            if thre_func.get_result():
                return thre_func.get_result()
            else:
                return"請求超時"  #超時返回  可以自定義

        return run

    return functions

#接口函數
def a1():
    print("開始請求接口")

    #這里把邏輯封裝成一個函數,使用線程調用
    a_theadiing = MyThread(target=a2)
    a_theadiing.start()
    a_theadiing.join()

    #返回結果
    a = a_theadiing.get_result()

    print("請求完成")
    return a
@limit_decor(3)   #超時設置為3s   2s邏輯未執行完畢返回接口超時
def a2():
    print("開始執行")
    time.sleep(2)
    print("執行完成")
    a=2
    return a

# 程序入口     未超時返回a的值   超時返回請求超時
if __name__ == '__main__':
    a = a1()  #調用接口(這里把函數a1看做一個接口)
    print(a)

超時設置3s，線程調用函數運行2s，這里返回a的值2。

 

方法2:使用信號模塊signal(只能在unix系統使用)

signal負責在Python程序內部處理信號，典型的操作包括預設信號處理函數，暫停並等待信號，以及定時發出SIGALRM等。

要注意，signal包主要是針對UNIX平台(比如Linux, MAC OS)，而Windows內核中由於對信號機制的支持不充分，所以在Windows上的Python不能發揮信號系統的功能。

信號是進程之間通訊的方式，是一種軟件中斷。一個進程一旦接收到信號就會打斷原來的程序執行流程來處理信號。

def set_timeout(num):
    def wrap(func):
        def handle(signum, frame):  # 收到信號 SIGALRM 后的回調函數，第一個參數是信號的數字，第二個參數是the interrupted stack frame.
            raise RuntimeError

        def to_do(*args):
            try:
                signal.signal(signal.SIGALRM, handle)  # 設置信號和回調函數
                signal.alarm(num)  # 設置 num 秒的鬧鍾
                print('start alarm signal.')
                r = func(*args)
                print('close alarm signal.')
                signal.alarm(0)  # 關閉鬧鍾
                return r
            except RuntimeError as e:
                return "超時啦"
        return to_do

    return wrap


@set_timeout(2)  # 限時 2 秒超時
def connect():  # 要執行的函數
    time.sleep(3)  # 函數執行時間，寫大於2的值，可測試超時
    return "完成"


if __name__ == '__main__':
    a = connect()