Python并行处理时无法按序保存LDAP用户层级数据问题

你的代码存在三个核心问题，导致无法获取完整的用户层级数据：

1. 多进程共享列表无效 #

多进程拥有独立的内存空间，你传递给run_parallel的self.users_ntid_list是主进程列表的副本，子进程对它的修改不会同步到主进程。而且多个进程同时往这个列表里写数据，还会导致数据混乱。

2. 函数无返回结果 #

run_parallel函数没有返回任何数据，pool.apply_async的get()方法只能拿到None，主进程自然收集不到任何层级信息。

3. LDAP层级查询逻辑错误 #

循环中重复查询同一个user，没有更新为经理的NTID，导致无法递归获取上级层级，甚至可能陷入无限循环。

修复后的代码 #

from datetime import datetime, timezone, timedelta
from time import sleep
from services.ldap import LDAP
import aiohttp
import os

from conf.settings import Settings as settings
import multiprocessing
from services import mongo
import asyncio


def run_parallel(user):
    # 每个进程创建独立的LDAP连接
    ldap_server = os.environ.get("LDAPSERVER")
    ldap_username = os.environ.get("SERVICE_ACCOUNT_USERNAME")
    ldap_password = os.environ.get("SERVICE_ACCOUNT_PASSWORD")
    ldap_base_dn = os.environ.get("BASE_DN")
    ldap_users = LDAP(ldap_server, ldap_username, ldap_password, ldap_base_dn)
    
    settings.log.info(f"Search Begins for User---->{user}")
    user_hierarchy = [user]  # 初始化当前用户的层级列表
    
    users_metadata = ldap_users.search(user)
    # 循环直到当前用户的经理是自己（最高层级）
    while users_metadata['ntid'] != users_metadata["ldap_manager"]:
        manager_ntid = users_metadata["ldap_manager"]
        settings.log.info(manager_ntid)
        user_hierarchy.append(manager_ntid)
        # 查询经理的元数据，进入下一轮循环
        users_metadata = ldap_users.search(manager_ntid)
    
    # 最后添加最高层级的自己（因为循环结束时经理等于自己，需要补全）
    if users_metadata['ntid'] not in user_hierarchy:
        user_hierarchy.append(users_metadata['ntid'])
    
    settings.log.info(f"User Hierarchy for {user}: {user_hierarchy}")
    return user_hierarchy  # 返回单个用户的层级列表


class LDAPUserSearch:
    def __init__(self, users_ntid_list):
        self.users_ntid_list = users_ntid_list 
        self.final_hierarchy = self.run()

    def run(self):
        users_ntid_list = ["rohan", "patel", "nancy", "umar", "john"]
        # 创建进程池
        with multiprocessing.Pool(processes=5) as pool:
            # 提交所有任务，获取结果列表
            results = pool.map(run_parallel, users_ntid_list)
        
        # 合并所有用户的层级列表
        final_hierarchy = []
        for hierarchy in results:
            final_hierarchy.extend(hierarchy)
        
        settings.log.info(f"Overall Complete Hierarchy: {final_hierarchy}")
        return final_hierarchy

关键修改点说明 #

• 移除共享列表：每个子进程独立生成单个用户的层级列表，通过返回值传递给主进程，避免多进程数据冲突。
• 修复查询逻辑：每次循环查询经理的元数据，确保层级递归正确，最后补全最高层级的节点。
• 使用pool.map简化代码：map方法可以直接迭代用户列表，自动收集所有返回结果，比apply_async更简洁。
• 用with管理进程池：自动处理池的关闭和回收，无需手动调用close()、join()。

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