限制函数在规定时间内只能执行一次-Toy模板网

这篇具有很好参考价值的文章主要介绍了限制函数在规定时间内只能执行一次。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

import time
from datetime import datetime
from datetime import timedelta


def timeguard(time_interval, default=None):
    def decorator(function):
        # For first time always run the function
        function.__last_run = datetime.min
        def guard(*args, **kwargs):
            now = datetime.now()
            if now - function.__last_run >= time_interval:
                function.__last_run = now
                return function(*args, **kwargs)
            elif default is not None:
                return default(*args, **kwargs)
        return guard
    return decorator


if __name__ == "__main__":
    
    @timeguard(timedelta(seconds=4),None)
    def add():
        print 12

    add()
    add()
    add()

记录一下，听常用的功能

def timeguard(time_interval, default=None):
    def decorator(function):
        # For first time always run the function
        function.__last_run = datetime.min
        def guard(*args, **kwargs):
            now = datetime.now()
            if now - function.__last_run >= time_interval:
                function.__last_run = now
                return function(*args, **kwargs)
            elif default is not None:
                return default(*args, **kwargs)
        return guard
    return decorator
@timeguard(timedelta(seconds=6),None)
def send_kafka_info(output_image_frame,alarm_content):

    bootstrap_servers = ['172.18.43.131:9092']
    topic = 'alg_recognition'
    
   
    # 上传minio
    # 发送kafka消息
    # 年月日分时秒的时间
    now_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    # 年月日时间
    data_time = now_time[0:10]
    kafka_info = {
        "businessCode": "zjk-env-sqgdj",
        "businessName": "张家口",
        "deviceNo": "show_room_person_count",
        "deviceName": "人员计数-展厅摄像头",
        "alarmType": "人员计数",
        "alarmTime": now_time,
        "alarmDate": data_time,
        "algCode": "ce202111180100007-zjk-env-sqgdj",
        "sourceImgUrl": "",
        "handleImgUrl": "",
        "videoUrl": "",
        "channel": "",
        "alarmContent": "",
        "algImageCode": "ce202111180100007",
        "trig": "周期检测优先"
    }

   

    
    try:
        # 写入图片
        img_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),"test_img",uuid.uuid4().hex + ".jpg")
        cv2.imwrite(img_path,output_image_frame)
        # 上传minio
        pic_url = save_picture_and_get_url_gevent(img_path)
        kafka_info['sourceImgUrl'] = pic_url
        kafka_info['handleImgUrl'] = pic_url
        kafka_info['alarmContent'] = alarm_content
        print({"msg": "通过kafka发送消息", "kafka_info": kafka_info})
        """
        {'msg': '通过kafka发送消息', 'kafka_info': {'businessCode': 'zjk-env-sqgdj', 'businessName': '张家口', 'deviceNo': 'show_room_person_count', 'deviceName': '人员计数-展厅摄像头', 'alarmType': '人员计数', 'alarmTime': '2023-07-07 11:34:21', 'alarmDate': '2023-07-07', 'algCode': 'ce202111180100007-zjk-env-sqgdj', 'sourceImgUrl': 'http://106.112.143.254:8333/bigscreen/alg_recognition/9c646c488d434b12bbe03f62967fb444.jpg', 'handleImgUrl': 'http://106.112.143.254:8333/bigscreen/alg_recognition/9c646c488d434b12bbe03f62967fb444.jpg', 'videoUrl': '', 'channel': '', 'alarmContent': '上行:0,下行:0', 'algImageCode': 'ce202111180100007', 'trig': '周期检测优先'}}
        """
        producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
        future = producer.send(topic, key=None, value=json.dumps(kafka_info, ensure_ascii=False).encode())
        #  kafka 10秒钟发不出去，就太过分了
        result = future.get(timeout=10)
        print("{} kafka res:".format(datetime.now()), result)
    except:
        print(datetime.now().strftime("%Y-%m-%f %H:%M:%s"), traceback.format_exc())