跨Python脚本进程互斥锁失效求助：locket库未按预期工作

问题根源 #

• test_lock_start.py执行后自动释放锁：脚本打印完内容就直接退出，操作系统会自动关闭进程持有的文件句柄，导致locket的文件锁被释放，所以test_lock_end.py能顺利拿到锁。
• test_lock_end.py未设置超时：locket.lock_file()默认无超时，acquire()会一直阻塞到拿到锁，不会抛出异常，你的try-except根本捕获不到错误。
• finally块的release逻辑有问题：如果没成功获取锁就调用release()，会直接触发异常。

修正后的代码 #

test_lock_start.py #

让进程保持运行，模拟持续持有锁的状态：

import locket
import os
import time

lock_path = os.path.join("./", "test_slot.lock")
test_lock = locket.lock_file(lock_path, timeout=2)
test_lock.acquire()
print("Ryan acquired Lock")
print("Ryan is dancing around")

# 保持进程运行，模拟长时间持有锁
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("\nRyan is stopping dance")
finally:
    test_lock.release()
    print("Ryan released Lock")

test_lock_end.py #

设置超时时间，仅在成功获取锁后执行逻辑，避免无锁释放的错误：

import locket
import os

lock_path = os.path.join("./", "test_slot.lock")
test_lock = locket.lock_file(lock_path, timeout=2)  # 设置超时时间

lock_acquired = False
try:
    print("Shanne trying to acquire lock")
    lock_acquired = test_lock.acquire()
    if lock_acquired:
        print("Shanne acquired lock")
        print("Shanne is dancing around")
    else:
        print("Unable to acquire lock")
        print("Shanne unable to dance")
except Exception as e:
    print(f"Error occurred: {e}")
    print("Shanne unable to dance")
finally:
    if lock_acquired:
        test_lock.release()
        print("Released Lock")
    print("Shanne can now try to acquire the lock later to dance")

运行验证 #

1. 先启动test_lock_start.py，脚本会打印内容并保持运行，此时锁处于被持有状态。
2. 再启动test_lock_end.py，会输出你预期的结果：

Shanne trying to acquire lock
Unable to acquire lock
Shanne unable to dance
Shanne can now try to acquire the lock later to dance

3. 按Ctrl+C停止test_lock_start.py，锁会被释放，此时再运行test_lock_end.py就能正常获取锁并执行后续逻辑。

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