基于唯一标识的多进程目录Git操作锁实现求助

你遇到的核心问题是每个进程单独创建的Lock对象是进程本地的，完全独立，所以根本无法实现跨进程的互斥。下面给你两个更优的方案，完美解决这个问题：

方案一：用multiprocessing.Manager共享锁字典 #

multiprocessing.Manager可以创建跨进程共享的对象，我们用它来维护一个字典——键是你的资源标识（比如'a'、'b'），值是对应的Lock对象。这样所有进程都会访问同一个锁字典，同一标识的进程会拿到同一个锁，自然实现互斥。

代码示例 #

from multiprocessing import Lock, Process, Manager
import os
import time

def work_on(name: str, lock_dict):
    # 确保每个标识对应唯一的锁（原子性的获取或创建）
    lock = lock_dict.get(name)
    if not lock:
        lock = Lock()
        lock_dict[name] = lock
    with lock:
        print(int(time.time() % 100), os.getpid(), f"working on {name}")
        time.sleep(5)
        print(int(time.time() % 100), os.getpid(), f"done with {name}")

if __name__ == "__main__":
    res = ['a', 'b']
    processes = []
    # 创建跨进程共享的字典
    with Manager() as manager:
        lock_dict = manager.dict()
        for r in res:
            p1 = Process(target=work_on, args=(r, lock_dict))
            p1.start()
            p2 = Process(target=work_on, args=(r, lock_dict))
            p2.start()
            processes += [p1, p2]
        
        for p in processes:
            p.join()

效果说明 #

这个方案会让同一标识的进程串行执行，不同标识的进程并行执行，完全符合你的预期。典型输出类似：

67 48687 working on a
67 48689 working on b
72 48687 done with a
72 48689 done with b
72 48688 working on a
72 48690 working on b
77 48688 done with a
77 48690 done with b

方案二：用文件系统原子锁（高性能/跨场景） #

如果不想依赖Manager的进程通信开销，或者需要支持跨语言、跨进程的锁，可以用文件系统的原子锁机制。利用操作系统对文件锁的原子性支持，彻底解决手动写文件的竞态问题，性能也更优。

Unix/Linux 下用fcntl实现 #

import fcntl
import os
import time
from multiprocessing import Process

def work_on(name: str):
    lock_file_path = f"./.lock_{name}"
    # 打开锁文件（不存在则自动创建）
    with open(lock_file_path, 'w') as lock_file:
        try:
            # 加独占锁，阻塞直到拿到锁
            fcntl.flock(lock_file, fcntl.LOCK_EX)
            print(int(time.time() % 100), os.getpid(), f"working on {name}")
            time.sleep(5)
            print(int(time.time() % 100), os.getpid(), f"done with {name}")
        finally:
            # 释放锁（文件关闭时也会自动释放，这里显式写是为了清晰）
            fcntl.flock(lock_file, fcntl.LOCK_UN)

if __name__ == "__main__":
    res = ['a', 'b']
    processes = []
    for r in res:
        p1 = Process(target=work_on, args=(r,))
        p1.start()
        p2 = Process(target=work_on, args=(r,))
        p2.start()
        processes += [p1, p2]
    
    for p in processes:
        p.join()

Windows 下用标准库实现 #

如果需要支持Windows，可以用win32file模块的锁机制：

import win32file
import win32con
import os
import time
from multiprocessing import Process

def work_on(name: str):
    lock_file_path = f"./.lock_{name}"
    # 创建或打开文件，指定不共享模式
    handle = win32file.CreateFile(
        lock_file_path,
        win32con.GENERIC_READ | win32con.GENERIC_WRITE,
        0,  # 禁止其他进程共享访问
        None,
        win32con.OPEN_ALWAYS,
        win32con.FILE_ATTRIBUTE_NORMAL,
        None
    )
    try:
        # 加独占锁
        win32file.LockFileEx(handle, win32con.LOCKFILE_EXCLUSIVE_LOCK, 0, -0x10000, None)
        print(int(time.time() % 100), os.getpid(), f"working on {name}")
        time.sleep(5)
        print(int(time.time() % 100), os.getpid(), f"done with {name}")
    finally:
        # 释放锁并关闭文件句柄
        win32file.UnlockFileEx(handle, 0, -0x10000, None)
        win32file.CloseHandle(handle)

if __name__ == "__main__":
    res = ['a', 'b']
    processes = []
    for r in res:
        p1 = Process(target=work_on, args=(r,))
        p1.start()
        p2 = Process(target=work_on, args=(r,))
        p2.start()
        processes += [p1, p2]
    
    for p in processes:
        p.join()

方案对比 #

• Manager字典锁：纯Python标准库，跨平台，实现简单，适合纯Python多进程场景；缺点是依赖Manager进程，有一定的通信开销。
• 文件原子锁：性能更高，支持跨进程/跨语言，不需要依赖Python的multiprocessing机制；缺点是需要处理平台差异，Unix和Windows的API不同。

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