Python multiprocessing下random重复取值与全局变量异常咨询

我目前正在处理一项优化问题，以下为整理的简化复现代码（原始代码逻辑过于复杂，不适宜直接用于问题咨询，该简化版本已尽可能还原原始代码的核心运行逻辑）。

实现目的 #

需要在optimization函数中调用foo函数，但foo在部分复杂场景下运行耗时极长，因此参考Stack Overflow平台关于函数调用执行时长限制的实现方案，使用multiprocessing模块为foo函数设置执行超时限制，超时逻辑通过proc.join(iter_time)方法实现。

遇到的问题 #

• while循环执行过程中，每次生成的extra变量取值完全相同
• 全局列表lst的长度始终为1，即每次while循环迭代时，lst都处于空列表的初始状态，无法保留历史迭代写入的数据

初步猜测 #

推测问题原因可能为：每次创建新进程时随机数种子都会从头开始计数，且进程终止后垃圾回收机制会清理该进程占用的内存空间，导致列表写入的数据被清空。

咨询问题 #

1. 出现上述两个问题的根本原因是什么？
2. 如果不使用multiprocessing模块，是否有其他方案可以实现函数执行超时控制，同时保证能够生成不同的随机数？注：已尝试过func_timeout库，但存在无法解决的其他兼容问题。

问题复现代码 #

import random
import time
import multiprocessing

random.seed(123)
lst = []  # 用于记录数据的全局列表

def foo(epoch):
    extra = random.random()
    lst.append(epoch + extra)

def optimization(loop_time, iter_time):
    start = time.time()
    epoch = 0
    while time.time() <= start + loop_time:
        proc = multiprocessing.Process(target=foo, args=(epoch,))
        proc.start()
        proc.join(iter_time)
        if proc.is_alive():  # 进程未在时限内结束
            print("Time out!")
            proc.terminate()
        epoch += 1

if __name__ == '__main__':
    optimization(300, 2)

问题根本原因 #

两个问题的核心原因都是multiprocessing的进程隔离特性，和垃圾回收没有关系：

• 随机数重复：你在主进程中执行了random.seed(123)，无论multiprocessing使用fork（类Unix系统默认）还是spawn（Windows/macOS默认）启动模式，每个新启动的子进程拿到的随机数初始状态完全一致，第一次调用random.random()自然会返回完全相同的值。
• 全局列表无法留存数据：操作系统会给每个进程分配独立的内存空间，子进程内的lst只是主进程初始空列表的副本，子进程对lst的修改只会作用在自己的内存空间里，主进程的lst从头到尾都没有被改动过。子进程退出后，它自己内存里修改过的列表会被系统直接回收，根本不会同步回主进程。

替代超时实现方案 #

排除多进程和func_timeout的前提下，有两个可落地的实现方案，都在同进程内执行，不存在内存隔离、随机数状态不同步的问题：

方案1：threading线程 + 退出标记 #

线程和主进程共享内存空间与随机数状态，没有隔离问题。需要注意Python的线程无法被强制外部终止，所以要在foo的耗时逻辑中插入检查点，响应退出标记：

import random
import time
import threading

random.seed(123)
lst = []
stop_flag = False

def foo(epoch):
    global stop_flag
    # 模拟耗时计算逻辑，在循环中插入退出检查点
    for _ in range(10000000):
        if stop_flag:
            return
        # 插入你的实际业务逻辑
    extra = random.random()
    lst.append(epoch + extra)

def optimization(loop_time, iter_time):
    global stop_flag
    start = time.time()
    epoch = 0
    while time.time() <= start + loop_time:
        stop_flag = False
        t = threading.Thread(target=foo, args=(epoch,))
        t.start()
        t.join(iter_time)
        if t.is_alive():
            print("Time out!")
            stop_flag = True
            t.join()
        epoch += 1

if __name__ == '__main__':
    optimization(10, 2)
    print(lst)  # 可正常看到累积的数据，随机值不重复

方案2：signal信号实现（仅支持类Unix系统） #

利用系统SIGALRM信号实现超时，完全单进程运行，代码更简洁，不需要改动foo内部逻辑：

import random
import time
import signal

random.seed(123)
lst = []

class FuncTimeout(Exception):
    pass

def timeout_handler(signum, frame):
    raise FuncTimeout()

def foo(epoch):
    time.sleep(3) # 模拟耗时操作
    extra = random.random()
    lst.append(epoch + extra)

def optimization(loop_time, iter_time):
    start = time.time()
    epoch = 0
    signal.signal(signal.SIGALRM, timeout_handler)
    while time.time() <= start + loop_time:
        signal.alarm(iter_time) # 配置定时信号
        try:
            foo(epoch)
        except FuncTimeout:
            print("Time out!")
        finally:
            signal.alarm(0) # 清空定时信号
        epoch += 1

if __name__ == '__main__':
    optimization(10, 2)
    print(lst)

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