当前位置:首页 > python > 正文内容

Python控制函数运行时间

root3年前 (2021-09-18)python589

在线程中主动判断时间的运行长度 

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)


使用信号模块signal (推荐使用,略复杂但好用)

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()


参考文档

扫描二维码推送至手机访问。

版权声明:本文由一叶知秋发布,如需转载请注明出处。

本文链接:https://zhiqiu.top/?id=147

分享给朋友:

相关文章

python 连接rabbitmq出现的诡异进程盗取消息

python 连接rabbitmq出现的诡异进程盗取消息

本文在py2下执行,由来:    因业务需要,python创建子线程后再次创建子进程(用于执行shell命令)。没错就是在子线程里面创建子进程。都知道py2的坑还是蛮多的。问题出现:在某次运行中出现了...

python 使用grpc 的方法

需要的依赖包grpcio googleapis-common-protos首先需要根据proto 文件生成代码proto 示例代码syntax = "proto3"; service ...

雪花算法 Snowflake python 实现代码

import time import logging # 分配位置 WORKER_BITS = 5 DATACENTER_BITS = 5 SEQUENCE...

python 发送和发送ICMP数据包

python 发送和发送ICMP数据包

ICMP协议在实际传输中数据包:20字节IP首部 + 8字节ICMP首部+ 1472字节<数据大小>38字节。对于ICMP首部细分为8位类型+8位代码+16位校验和+16位标识符+16位序列号其中类型的取值如下,我们比较关注的是...

Python  os.system 和subprocess.popen 并发执行linux的性能对比

Python os.system 和subprocess.popen 并发执行linux的性能对比

os subprocess multiprocessing.dummy Pool ThreadPool command_list [] ()os.() start_time =&...

python 实现AES加密解密

什么是非对称加密1. A要向B发送信息,A和B都要产生一对用于加密和解密的公钥和私钥。• 2. A的私钥保密,A的公钥告诉B;B的私钥保密,B的公钥告诉A。• 3. A要给B发送信息时,A用B的公钥加密信息,因为A知道B的公钥。• 4. A...