Python多进程下如何实现createIndexForCollection函数的互斥访问

你的需求是保证同一时间只有一个进程能执行createIndexForCollection函数，其他进程必须等待当前进程执行完该函数后才能进入。因为线程锁无法跨进程生效，所以需要使用进程间同步锁来实现这个逻辑。

实现思路 #

使用multiprocessing.Manager创建一个跨进程共享的锁对象，每个启动的进程都持有这个锁的引用。在调用createIndexForCollection前先获取锁，执行完成后释放锁，这样就能保证同一时间只有一个进程进入该函数的执行流程。

修改后的完整代码 #

import multiprocessing as mp
import threading

def createSigneIndexForCollection(mongoDatabase, collectionName, columnName):
    rawlogCollection = mongoDatabase[collectionName]
    rawlogCollection.create_index([(columnName , 1)])

def createIndexForCollection(mongoDatabase, collectionName):
    columnsToBeIndexed = ["Name", "Message"]  # 修正原代码语法错误，实际使用请补充完整列名
    for columnName in columnsToBeIndexed:
        indexThread = threading.Thread(target=createSigneIndexForCollection, args=(mongoDatabase, collectionName, columnName))
        indexThread.start()
        indexThread.join()

def a_process(processCollection, mongoIP, mongoPort, index_lock):
    # 自动获取/释放锁，确保进程间互斥执行索引创建逻辑
    with index_lock:
        mongoDatabase = getDatabase(mongoIP, mongoPort, mongoDatabase)  # 假设getDatabase为已实现的数据库连接函数
        createIndexForCollection(mongoDatabase, processCollection)
     
    if processCollection != "NULL":
        ScanningThread = threading.Thread(target=check, args=(mongoDatabase, processCollection))
        ScanningThread.start()
        ScanningThread.join()

if __name__ == "__main__":
    # 创建跨进程共享的锁实例
    with mp.Manager() as manager:
        index_lock = manager.Lock()
        # 示例：启动单个进程（实际每6分钟启动的逻辑需结合定时调度框架实现，如APScheduler）
        processCollection = "your_target_collection"
        mongoIP = "127.0.0.1"
        mongoPort = 27017
        processingP = mp.Process(target=a_process, args=(processCollection, mongoIP, mongoPort, index_lock))
        processingP.start()

关键说明 #

• mp.Manager().Lock()创建的锁可在不同进程间共享，是实现进程互斥的核心。
• with index_lock:语句会自动处理锁的获取与释放，避免手动操作锁时可能出现的死锁问题。
• 原代码中columnsToBeIndexed的写法存在语法错误，已修正为合法列表格式，实际使用时请补充全需要建立索引的列名。

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