Python 通用锁机制-支持同时管理多个锁

from threading import Lock
from typing import Any, Dict, Set, List

class MultiLockManager:
    def __init__(self):
        self._global_lock = Lock()  # 用于保护资源类别的访问
        self._resource_locks: Dict[str, Set[Any]] = {}

    def acquire(self, category: str, resource: Any) -> bool:
        """尝试获取单个资源锁"""
        with self._global_lock:
            if category not in self._resource_locks:
                self._resource_locks[category] = set()
            if resource in self._resource_locks[category]:
                return False
            self._resource_locks[category].add(resource)
            return True

    def release(self, category: str, resource: Any):
        """释放单个资源锁"""
        with self._global_lock:
            if category in self._resource_locks:
                self._resource_locks[category].discard(resource)

    def acquire_batch(self, category: str, resources: List[Any]) -> bool:
        """
        尝试锁定多个资源，全部成功才锁定，否则回滚
        :param category: 资源类别
        :param resources: 要锁定的资源列表
        :return: True 如果全部资源成功锁定，否则 False
        """
        locked_resources = []
        with self._global_lock:
            if category not in self._resource_locks:
                self._resource_locks[category] = set()
            for resource in resources:
                if resource in self._resource_locks[category]:
                    # 回滚已锁定的资源
                    for locked in locked_resources:
                        self._resource_locks[category].discard(locked)
                    return False
                self._resource_locks[category].add(resource)
                locked_resources.append(resource)
            return True

    def release_batch(self, category: str, resources: List[Any]):
        """释放多个资源锁"""
        with self._global_lock:
            if category in self._resource_locks:
                for resource in resources:
                    self._resource_locks[category].discard(resource)

    def is_locked(self, category: str, resource: Any) -> bool:
        """检查单个资源是否已被锁定"""
        with self._global_lock:
            return category in self._resource_locks and resource in self._resource_locks[category]

更新：不会用ChatGPT的程序员不是好程序员（笑）