基于Redis自动过期的流处理暂停机制

class tracking_car:
    def __init__(self, profile_path, logger_) -> none:
        # ...其他初始化代码...
        
        # redis连接
        self.redis_db = redis.strictredis(
            host=conf.redis_server.ip,
            port=conf.redis_server.port,
            db=conf.redis_server.db,
            socket_keepalive=true,
            socket_connect_timeout=10
        )
        
        # 超时存储的redis key前缀
        self.timeout_key_prefix = "tracking_car:timeout:"

    def re_stream(self, logger_):
        pub = self.redis_db.pubsub()
        pub.subscribe(self.topic)
        msgs = pub.listen()
        
        for msg in msgs:
            if msg["type"] == "message":
                json_data = json.loads(msg["data"])
                ip = json_data["ip"]
                
                # 检查是否在暂停列表 - 使用redis自动过期
                timeout_key = f"{self.timeout_key_prefix}{camera_ip}"
                if self.redis_db.exists(timeout_key):
                    # 获取剩余时间并记录日志
                    ttl = self.redis_db.ttl(timeout_key)
                    skip_msg = f"跳过{ip }的消息：处于暂停时段({ttl}s剩余)"
                    continue
                
                # ...正常处理逻辑...

    def write_database(self, cv_list, logger_: mylogger):
        # 计算时间差
        current_time = time.time()
        _ts = cv_list['timestamp']
        diff_time = current_time - _ts
        
        # 如果时间差超过300秒，使用redis自动过期设置
        if diff_time > 300:
            camera_ip = cv_list["ip"]
            logger_.warning(f"ip {camera_ip} 延时超过300秒({diff_time:.2f}s)，加入暂停列表")
            
            # 设置redis键，自动在300秒后过期
            timeout_key = f"{self.timeout_key_prefix}{camera_ip}"
            self.redis_db.setex(timeout_key, 300, "1")  # 值可以是任意内容
            
            # 删除相关图片并跳过处理
            self.redis_db.unlink(cv_list["path"])
            return
        
        # ...正常处理逻辑...