python lock锁 （装饰器）

#1.普通全局锁

# 2. 支持不同方法的独立锁  (必须有该方法)

 

1.普通全局锁

from functools import wraps
from typing import Dict, Any, List, Optional
import uuid
import tempfile
import shutil
import re
import time# 历史记录文件
HISTORY_FILE="./history/history.json"
# 全局锁，避免多线程冲突
GLOBAL_LOCK = threading.Lock()
def synchronized_instance_method_with_retry(max_retries=10, delay=0.3, backoff=2, timeout=30):"""
    高性能版本：针对文件I/O操作优化- 优先使用简单锁机制- 只在需要时才启用超时监控- 减少线程创建频率"""
    def decorator(func):@wraps(func)def wrapper(*args, **kwargs):last_exception = Nonefor attempt in range(max_retries + 1):try:with GLOBAL_LOCK:start_time = time.time()# 对于文件操作，通常不需要复杂的超时机制# 因为文件操作通常是同步的result = func(*args, **kwargs)# 如果执行时间超过超时阈值，发出警告execution_time = time.time() - start_timeif execution_time > timeout:print(f"Warning: Function took {execution_time:.2f}s (threshold: {timeout}s)")return resultexcept Exception as e:last_exception = eprint(f"Attempt {attempt + 1} of {max_retries + 1} failed: {e}")if attempt == max_retries:raise last_exceptiontime.sleep(delay * (backoff ** attempt))return Nonereturn wrapperreturn decorator@synchronized_instance_method_with_retry()
def save(history={},history_file=HISTORY_FILE):"""
    保存历史记录 把文件写入 本地json文件，可以一直追加，后续可以查询

 

 

 

 

 

 

# 2. 支持不同方法的独立锁  (必须有该方法)

# 支持不同方法的独立锁
def synchronized_instance_method_with_retry(method_name="add", max_retries=30,timeout=1.0, retry_delay=0.1):"""
    为每个方法创建独立的实例锁，支持超时重试参数:method_name: 方法名，用于生成锁属性名max_retries: 最大重试次数timeout: 每次获取锁的超时时间（秒）retry_delay: 重试延迟时间（秒）"""
    def decorator(func):@wraps(func)def wrapper(self, *args, **kwargs):# 生成锁属性名if method_name:lock_attr = f"_lock_{method_name}"else:lock_attr = f"_lock_{func.__name__}"# 确保实例有该方法的锁if not hasattr(self, lock_attr):lock = threading.Lock()setattr(self, lock_attr, lock)lock = getattr(self, lock_attr)retries = 0last_exception = Nonewhile retries <= max_retries:acquired = Falsetry:# 使用退避算法获取锁acquired = lock.acquire(timeout=timeout)if acquired:return func(self, *args, **kwargs)else:retries += 1if retries <= max_retries:# 指数退避策略wait_time = retry_delay * (2 ** retries)  # 指数增加wait_time = min(wait_time, 5.0)  # 上限5秒print(f"[WARNING] {self.__class__.__name__}.{func.__name__}: "f"第 {retries} 次重试，等待 {wait_time:.2f}s...")time.sleep(wait_time)except Exception as e:last_exception = eif acquired:lock.release()raise efinally:if acquired:lock.release()error_msg = f"{self.__class__.__name__}.{func.__name__}: 重试 {max_retries} 次后仍失败"print(f"[ERROR] {error_msg}")raise TimeoutError(error_msg) from last_exceptionreturn wrapperreturn decorator

使用：

# 支持不同方法的独立锁
def synchronized_instance_method_with_retry(method_name="add", max_retries=30,timeout=1.0, retry_delay=0.1):"""
    为每个方法创建独立的实例锁，支持超时重试参数:method_name: 方法名，用于生成锁属性名max_retries: 最大重试次数timeout: 每次获取锁的超时时间（秒）retry_delay: 重试延迟时间（秒）"""
    def decorator(func):passclass HistoryUtils:def __init__(self, history_file="./history/history.json"):#历史记录文件self.history_file = history_file   @synchronized_instance_method_with_retry(method_name="save")def save(self, history={},history_file= None):"""
        保存历史记录 把文件写入 本地json文件，可以一直追加，后续可以查询:param history_file: 历史记录文件 eg:./history/history.json:param history: 历史记录 eg:{"name": "张三9", "age": 27}