为何在Python多进程中全局列表无法被追加？

没错，就是多进程导致的。Python多进程机制中，每个子进程会拥有独立的内存空间，主进程的全局变量会被复制一份到子进程——你在子进程里修改的uncounted只是它自己的副本，完全不会影响主进程里的原列表，所以最后主进程打印出来的还是空的。

要让子进程的修改同步到主进程，必须用multiprocessing模块提供的进程间通信（IPC）工具，比如Manager创建的共享列表，或者Queue队列。

方案1：用Manager创建共享列表 #

这是最贴近你原有代码逻辑的修改方式，只需把普通列表换成Manager管理的共享列表：

import multiprocessing
import os
import time

# 关键修改1：用Manager创建可跨进程共享的列表
manager = multiprocessing.Manager()
uncounted = manager.list()
test = [{'id': 'UC8YCf2zyqY8E7-CCOnr_kQA', 'name': 'dominic alting:('}, {'id': 'UCxm1n3Hil3l9mmPWcONd4Dg', 'name': 'Toes Jr.'}]

dir_path = "./json/"

def load(pathD):
    # 原代码中pathD参数未被使用，可按需删除或调整逻辑
    return test

def multiprocessing_func(dp, shared_list):
    # 关键修改2：直接传递共享列表，无需global关键字
    time.sleep(2)
    u = load(dp)
    shared_list.append(u)

errors = []

def main():
    if __name__ == '__main__':
        starttime = time.time()
        processes = []
        # 提前获取目录文件列表，避免重复调用且防止索引越界
        dir_files = os.listdir(dir_path)
        # 限制进程数不超过文件总数
        max_processes = min(10, len(dir_files))
        
        for i in range(max_processes):
            print(f'{i} has started')
            try:
                # 关键修改3：将共享列表作为参数传给子进程
                p = multiprocessing.Process(
                    target=multiprocessing_func, args=(dir_files[i], uncounted))
                processes.append(p)
                p.start()
            except Exception as e:
                errors.append(i)
                print(f"{i} - 出错：{str(e)}")

        for process in processes:
            process.join()
        
        print(f'耗时：{time.time() - starttime}秒')
        print(uncounted)

main()

方案2：用Queue收集结果 #

如果不需要实时共享列表，也可以用队列让子进程把结果传递回来，主进程最后统一整理：

import multiprocessing
import os
import time

test = [{'id': 'UC8YCf2zyqY8E7-CCOnr_kQA', 'name': 'dominic alting:('}, {'id': 'UCxm1n3Hil3l9mmPWcONd4Dg', 'name': 'Toes Jr.'}]

dir_path = "./json/"

def load(pathD):
    return test

def multiprocessing_func(dp, queue):
    time.sleep(2)
    u = load(dp)
    queue.put(u)

errors = []

def main():
    if __name__ == '__main__':
        starttime = time.time()
        queue = multiprocessing.Queue()
        processes = []
        dir_files = os.listdir(dir_path)
        max_processes = min(10, len(dir_files))
        
        for i in range(max_processes):
            print(f'{i} has started')
            try:
                p = multiprocessing.Process(
                    target=multiprocessing_func, args=(dir_files[i], queue))
                processes.append(p)
                p.start()
            except Exception as e:
                errors.append(i)
                print(f"{i} - 出错：{str(e)}")

        for process in processes:
            process.join()
        
        # 从队列中取出所有结果
        uncounted = []
        while not queue.empty():
            uncounted.append(queue.get())
        
        print(f'耗时：{time.time() - starttime}秒')
        print(uncounted)

main()

• load函数的pathD参数被直接赋值为test，完全没用到传入的参数，建议按需删除或调整逻辑
• 原循环里的i += 1是多余的，range(0,10)会自动迭代i的值，这会导致索引混乱
• 直接用os.listdir(dir_path)[xi]可能因目录文件数不足10个触发索引越界，建议先获取文件列表再做判断

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