Python中是否存在带密钥的同步锁，仅允许持有锁的进程释放？

你完全不需要自定义Manager子类并注册自定义类，用标准库默认的multiprocessing.Manager提供的共享对象封装即可，代码量可以缩减一半以上：

方案1：函数式封装（最简实现） #

直接用Manager创建共享锁和持有者标识，用工具函数完成加锁释放逻辑，不需要定义类：

from multiprocessing import Manager
import os

# 初始化共享资源，全局仅执行一次
manager = Manager()
_base_lock = manager.Lock()
_lock_holder = manager.Value('i', 0)  # 存储持有锁的进程PID，0代表锁未被持有

def acquire_lock(blocking: bool = True, timeout: float = 10.0) -> bool:
    if _base_lock.acquire(blocking=blocking, timeout=timeout):
        _lock_holder.value = os.getpid()
        return True
    return False

def release_lock(raise_error: bool = False) -> bool:
    if _lock_holder.value == os.getpid():
        _lock_holder.value = 0
        _base_lock.release()
        return True
    if raise_error:
        raise RuntimeError('仅锁的持有进程可执行释放操作')
    return False

调用方式和你之前的用法基本一致：

acquire_lock()
# 操作共享资源
...
release_lock()

方案2：面向对象封装（保留类使用习惯） #

如果你更习惯类的调用方式，也可以直接把共享对象作为初始化参数传入普通类，不需要向Manager注册自定义类：

from multiprocessing import Manager, Lock
from multiprocessing.managers import ValueProxy
import os

class KeyLock:
    def __init__(self, base_lock: Lock, holder: ValueProxy):
        self._lock = base_lock
        self._holder = holder
    
    def acquire(self, blocking: bool = True, timeout: float = 10.0) -> bool:
        if self._lock.acquire(blocking=blocking, timeout=timeout):
            self._holder.value = os.getpid()
            return True
        return False
    
    def release(self, raise_error: bool = False) -> bool:
        if self._holder.value == os.getpid():
            self._holder.value = 0
            self._lock.release()
            return True
        if raise_error:
            raise RuntimeError('仅锁的持有进程可执行释放操作')
        return False

# 初始化锁实例，全局仅执行一次
manager = Manager()
lock = KeyLock(manager.Lock(), manager.Value('i', 0))

使用方式和你原来的实现完全一致，不需要修改现有业务代码。

两种方案都避免了自定义Manager子类、注册类的冗余代码，完全满足「仅持有锁的进程可释放锁」的需求，同时兼容所有支持multiprocessing的操作系统。

内容的提问来源于stack exchange，提问作者Stefan Meier